feat: add table with ingoing transactions (#23)

aggregations/__init__.py

@@ -3,6 +3,7 @@
 from .db_tables.daily_deleted_accounts_count import DailyDeletedAccountsCount
 from .db_tables.daily_deposit_amount import DailyDepositAmount
 from .db_tables.daily_gas_used import DailyGasUsed
+from .db_tables.daily_ingoing_transactions_per_account_count import DailyIngoingTransactionsPerAccountCount
 from .db_tables.daily_new_accounts_count import DailyNewAccountsCount
 from .db_tables.daily_new_contracts_count import DailyNewContractsCount
 from .db_tables.daily_new_unique_contracts_count import DailyNewUniqueContractsCount

aggregations/db_tables/daily_ingoing_transactions_per_account_count.py

@@ -0,0 +1,78 @@
+import datetime
+
+from . import DAY_LEN_SECONDS, daily_start_of_range
+from ..periodic_aggregations import PeriodicAggregations
+
+
+class DailyIngoingTransactionsPerAccountCount(PeriodicAggregations):
+    @property
+    def sql_create_table(self):
+        # Suppose we have at most 10^5 (100K) transactions per second.
+        # In the worst case, they are all from one account.
+        # It gives ~10^10 transactions per day.
+        # It means we fit into BIGINT (10^18)
+        return '''
+            CREATE TABLE IF NOT EXISTS daily_ingoing_transactions_per_account_count
+            (
+                collected_for_day          DATE   NOT NULL,
+                account_id                 TEXT   NOT NULL,
+                ingoing_transactions_count BIGINT NOT NULL,
+                CONSTRAINT daily_ingoing_transactions_per_account_count_pk PRIMARY KEY (collected_for_day, account_id)
+            );
+            CREATE INDEX IF NOT EXISTS daily_ingoing_transactions_per_account_count_idx
+                ON daily_ingoing_transactions_per_account_count (collected_for_day, ingoing_transactions_count DESC)
+        '''
+
+    @property
+    def sql_drop_table(self):
+        return '''
+            DROP TABLE IF EXISTS daily_ingoing_transactions_per_account_count
+        '''
+
+    @property
+    def sql_select(self):
+        # Ingoing transactions for user X aren't only transactions where receiver_account_id == X.
+        # We need to find all chains with receipts where X was the receiver.
+        # It's important to add 10 minutes to receipt border, we should pack the transaction
+        # with all their receipts together, or the numbers will not be accurate.
+        # Other receipts from the next day will be naturally ignored.
+        # Transactions border remains the same, taking only transactions for the specified day.
+        # If you want to change 10 minutes constant, fix it also in PeriodicAggregations.is_indexer_ready
+
+        # Conditions on receipts timestamps are added because of performance issues:
+        # Joining 2 relatively small tables work much faster (4-6s VS 70-150s)
+        # Conditions on transactions timestamps are required by design.
+        # Though, they were placed into JOIN section also because of performance issues. Not sure why,
+        # but it changes the query plan to a better one and gives much better performance
+        return '''
+            SELECT
+                receipts.receiver_account_id,
+                COUNT(DISTINCT transactions.transaction_hash) AS ingoing_transactions_count
+            FROM transactions
+            LEFT JOIN receipts ON receipts.originated_from_transaction_hash = transactions.transaction_hash
+                AND transactions.block_timestamp >= %(from_timestamp)s
+                AND transactions.block_timestamp < %(to_timestamp)s
+            WHERE receipts.included_in_block_timestamp >= %(from_timestamp)s
+                AND receipts.included_in_block_timestamp < (%(to_timestamp)s + 600000000000)
+                AND transactions.signer_account_id != receipts.receiver_account_id 
+            GROUP BY receipts.receiver_account_id
+        '''
+
+    @property
+    def sql_insert(self):
+        return '''
+            INSERT INTO daily_ingoing_transactions_per_account_count VALUES %s
+            ON CONFLICT DO NOTHING
+        '''
+
+    @property
+    def duration_seconds(self):
+        return DAY_LEN_SECONDS
+
+    def start_of_range(self, timestamp: int) -> int:
+        return daily_start_of_range(timestamp)
+
+    @staticmethod
+    def prepare_data(parameters: list, *, start_of_range=None, **kwargs) -> list:
+        computed_for = datetime.datetime.utcfromtimestamp(start_of_range).strftime('%Y-%m-%d')
+        return [(computed_for, account_id, count) for (account_id, count) in parameters]

aggregations/db_tables/daily_outgoing_transactions_per_account_count.py

@@ -56,6 +56,6 @@ def start_of_range(self, timestamp: int) -> int:
         return daily_start_of_range(timestamp)
 
     @staticmethod
-    def prepare_data(parameters: list, **kwargs) -> list:
-        computed_for = datetime.datetime.utcfromtimestamp(kwargs['start_of_range']).strftime('%Y-%m-%d')
+    def prepare_data(parameters: list, *, start_of_range=None, **kwargs) -> list:
+        computed_for = datetime.datetime.utcfromtimestamp(start_of_range).strftime('%Y-%m-%d')
         return [(computed_for, account_id, count) for (account_id, count) in parameters]

aggregations/db_tables/daily_receipts_per_contract_count.py

@@ -57,6 +57,6 @@ def start_of_range(self, timestamp: int) -> int:
         return daily_start_of_range(timestamp)
 
     @staticmethod
-    def prepare_data(parameters: list, **kwargs) -> list:
-        computed_for = datetime.datetime.utcfromtimestamp(kwargs['start_of_range']).strftime('%Y-%m-%d')
+    def prepare_data(parameters: list, *, start_of_range=None, **kwargs) -> list:
+        computed_for = datetime.datetime.utcfromtimestamp(start_of_range).strftime('%Y-%m-%d')
         return [(computed_for, contract_id, count) for (contract_id, count) in parameters]

aggregations/periodic_aggregations.py

@@ -26,11 +26,11 @@ def collect(self, requested_timestamp: int) -> list:
             return self.prepare_data(result, start_of_range=from_timestamp)
 
     @staticmethod
-    def prepare_data(parameters: list, **kwargs) -> list:
+    def prepare_data(parameters: list, *, start_of_range=None, **kwargs) -> list:
         # We usually have one-value returns, we need to merge it with corresponding date
         if len(parameters[0]) == 1:
             assert len(parameters) == 1, 'Only one value expected. Can\'t be sure that we need to add timestamp'
-            computed_for = datetime.datetime.utcfromtimestamp(kwargs['start_of_range'])
+            computed_for = datetime.datetime.utcfromtimestamp(start_of_range)
             parameters = [(computed_for, parameters[0][0] or 0)]
         return [(computed_for.strftime('%Y-%m-%d'), data) for (computed_for, data) in parameters]
 
@@ -44,4 +44,6 @@ def is_indexer_ready(self, needed_timestamp):
         with self.indexer_connection.cursor() as indexer_cursor:
             indexer_cursor.execute(select_latest_timestamp)
             latest_timestamp = indexer_cursor.fetchone()[0]
-            return latest_timestamp >= needed_timestamp
+            # Adding 10 minutes to be sure that all the data is collected
+            # Important for DailyIngoingTransactionsPerAccountCount
+            return latest_timestamp >= needed_timestamp + 10 * 60

main.py

@@ -6,10 +6,10 @@
 import typing
 
 from aggregations import DailyActiveAccountsCount, DailyActiveContractsCount, DailyDeletedAccountsCount, \
-    DailyDepositAmount, DailyGasUsed, DailyNewAccountsCount, DailyNewContractsCount, DailyNewUniqueContractsCount, \
-    DailyOutgoingTransactionsPerAccountCount, DailyReceiptsPerContractCount, DailyTokensSpentOnFees, \
-    DailyTransactionsCount, DailyTransactionsPerAccountCount, DeployedContracts, WeeklyActiveAccountsCount, \
-    NearEcosystemEntities
+    DailyDepositAmount, DailyGasUsed, DailyIngoingTransactionsPerAccountCount, DailyNewAccountsCount, \
+    DailyNewContractsCount, DailyNewUniqueContractsCount, DailyOutgoingTransactionsPerAccountCount, \
+    DailyReceiptsPerContractCount, DailyTokensSpentOnFees, DailyTransactionsCount, DailyTransactionsPerAccountCount, \
+    DeployedContracts, WeeklyActiveAccountsCount, NearEcosystemEntities
 from aggregations.db_tables import DAY_LEN_SECONDS, query_genesis_timestamp
 
 from datetime import datetime
@@ -21,6 +21,7 @@
     'daily_deleted_accounts_count': DailyDeletedAccountsCount,
     'daily_deposit_amount': DailyDepositAmount,
     'daily_gas_used': DailyGasUsed,
+    'daily_ingoing_transactions_per_account_count': DailyIngoingTransactionsPerAccountCount,
     'daily_new_accounts_count': DailyNewAccountsCount,
     'daily_new_contracts_count': DailyNewContractsCount,
     'daily_new_unique_contracts_count': DailyNewUniqueContractsCount,