Merge pull request #53 from ton-society/hotfix/onchain_mau

Refactoring unique users export pipeline

backends/redoubt/apps.py

@@ -61,19 +61,20 @@ def _do_calculate(self, config: SeasonConfig, dry_run: bool = False):
             messages = f"""
             -- full messages table, filter by last 30 days
             select m.*
-            from  messages m
+            from  transactions t
+            join messages m on m.in_tx_id  = t.tx_id
+    
             where
-            (
-            select
+            t.utime >= {config.start_time}::int and t.utime < {config.end_time}::int
+            and (
                         (t.action_result_code  = 0 and t.compute_exit_code  = 0)
                         or
                         (t.action_result_code is null and t.compute_exit_code  is null and t.compute_skip_reason = 'cskip_no_gas')
-            from transactions t where t.tx_id = in_tx_id and t.utime > {config.start_time} and 
-            t.utime < {config.end_time} 
             )
             """
-            final_part = """
-            select count(1) as mau from users_stats where tx_count > 1
+            final_part = f"""
+            insert into tol.daily_app_users(project, user_address, tx_count, period_start, period_end)
+            select project, user_address, tx_count, {config.start_time}::int as period_start, {config.end_time}::int as period_end from users_stats
             """
         else:
             messages = f"""
@@ -107,37 +108,37 @@ def _do_calculate(self, config: SeasonConfig, dry_run: bool = False):
             JOIN jetton_wallets jw ON jw.address = jt.source_wallet and not jw.is_scam
             where
                 jt.successful and
-                jt.utime >= {config.start_time} and
-                jt.utime <  {config.end_time}
+                jt.utime >= {config.start_time}::int and
+                jt.utime <  {config.end_time}::int
         ), nft_activity_local as (
           select msg_id as id, nt.current_owner as user_address, ni.collection  from nft_transfers nt
                                                                                join nft_item ni on nt.nft_item = ni.address
-            where nt.utime >= {config.start_time}  and nt.utime <  {config.end_time}
+            where nt.utime >= {config.start_time}::int  and nt.utime <  {config.end_time}::int
               and collection is not null
             union
             select msg_id as id, new_owner as user_address, collection_address as collection
             from nft_history nh where event_type ='sale'
-                                  and utime >= {config.start_time}  and utime <  {config.end_time}
+                                  and utime >= {config.start_time}::int  and utime <  {config.end_time}::int
         ), nft_history_local as (
             select  * from nft_history
-            where utime  >= {config.start_time} and utime  < {config.end_time}
+            where utime  >= {config.start_time}::int and utime  < {config.end_time}::int
         ), nft_transfers_local as (
             select  * from nft_transfers
-            where utime  >= {config.start_time} and utime  < {config.end_time}
+            where utime  >= {config.start_time}::int and utime  < {config.end_time}::int
         ), ton20_sale_local as (
             select * from ton20_sale ts
-            where utime >= {config.start_time}  and utime <  {config.end_time}
+            where utime >= {config.start_time}::int  and utime <  {config.end_time}::int
         ), jetton_burn_local as (
             select jb.*, jw."owner" as user_address, jw.jetton_master from jetton_burn jb
             join jetton_wallets jw on jw.address  = jb.wallet and jb.successful and not jw.is_scam
-            where utime >= {config.start_time}  and utime <  {config.end_time}
+            where utime >= {config.start_time}::int  and utime <  {config.end_time}::int
         ), jetton_mint_local as (
             select jm.*, jw."owner" as user_address, jw.jetton_master from jetton_mint jm
             join jetton_wallets jw on jw.address  = jm.wallet and jm.successful and not jw.is_scam
-            where utime >= {config.start_time}  and utime <  {config.end_time}
+            where utime >= {config.start_time}::int  and utime <  {config.end_time}::int
         ), dex_swaps_local as (
             select * from dex_swap_parsed
-            where swap_utime >= {config.start_time}  and swap_utime <  {config.end_time}
+            where swap_utime >= {config.start_time}::int  and swap_utime <  {config.end_time}::int
         ),      
         nft_sales as (
             select msg_id as id, nh.current_owner  as user_address, marketplace from nft_history_local nh where
@@ -197,10 +198,10 @@ def _do_calculate(self, config: SeasonConfig, dry_run: bool = False):
 
         if self.mau_stats:
             with self.connection.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cursor:
+                logger.info("Executing insert")
                 cursor.execute(SQL)
-                mau = cursor.fetchone()['mau']
-                logger.info(f"Mau calculated: {mau}")
-                return mau
+                logger.info("Query finished")
+                return
         if dry_run:
             logger.info("Running SQL query in dry_run mode")
             with self.connection.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cursor:

backends/redoubt/nfts.py

@@ -48,8 +48,8 @@ def _do_calculate(self, config: SeasonConfig, dry_run: bool = False):
             from nft_history nh 
             join collections c on c.address = nh.collection_address
             where event_type = 'sale'
-            and utime > {config.start_time}
-            and utime < {config.end_time}
+            and utime > {config.start_time}::int
+            and utime < {config.end_time}::int
         ),
         top as (
             select d.collection_address address, sum(d.price) / 1e9 volume

models/backend.py

@@ -21,7 +21,7 @@ def calculate(self, config: SeasonConfig, dry_run: bool = False):
         if isinstance(res, CalculationResults):
             logger.info(f"Applying scoring model to {res}")
             res.ranking = config.score_model.calculate(res.ranking)
-        if not dry_run:
+        if not dry_run and res:
             res.build_time = update_time
         return res
 

seasons/__init__.py

@@ -1,17 +0,0 @@
-import time
-
-from models.season_config import SeasonConfig
-from seasons.s3_5 import S3_5_apps
-import time
-
-from seasons.app_models import AppLeaderboardModelV2
-
-start_time = int(time.time()) - 3 * 3600 # 3 hours to mitigate possible indexer lag
-LAST_30DAYS_MAU_SEASON = SeasonConfig(
-    leaderboard=SeasonConfig.APPS,
-    name="last",
-    start_time=start_time - 30 * 86400,
-    end_time=start_time,
-    projects=S3_5_apps.projects,
-    score_model=AppLeaderboardModelV2()
-)