[DRAFT] Build Enriched Traces & Transactions + Aggregation Tables - pt2

notebooks/adhoc/refined_traces_models_dev_v2.ipynb

@@ -4,7 +4,7 @@
    "cell_type": "markdown",
    "metadata": {},
    "source": [
-    "## Prepare data reader and up model execution context"
+    "## Prepare data reader and model execution context"
    ]
   },
   {

src/op_analytics/datapipeline/models/code/daily_address_summary_old.py

@@ -0,0 +1,21 @@
+# import duckdb
+
+# from op_analytics.datapipeline.models.compute.querybuilder import TemplatedSQLQuery
+# from op_analytics.datapipeline.models.compute.registry import register_model
+# from op_analytics.datapipeline.models.compute.types import NamedRelations
+
+
+# @register_model(
+#     input_datasets=["intermediate/refined_transactions_fees_v1"],
+#     expected_outputs=["summary_v1"],
+#     auxiliary_views=[
+#         TemplatedSQLQuery(
+#             template_name="daily_address_summary",
+#             context={},
+#         ),
+#     ],
+# )
+# def daily_address_summary_old(duckdb_client: duckdb.DuckDBPyConnection) -> NamedRelations:
+#     return {
+#         "summary_v1": duckdb_client.view("daily_address_summary"),
+#     }

src/op_analytics/datapipeline/models/code/daily_transactions_fees_by_to.py

@@ -0,0 +1,26 @@
+import duckdb
+
+from op_analytics.datapipeline.models.compute.querybuilder import TemplatedSQLQuery
+from op_analytics.datapipeline.models.compute.registry import register_model
+from op_analytics.datapipeline.models.compute.types import NamedRelations
+
+
+@register_model(
+    input_datasets=["intermediate/enriched_transactions_v1"],
+    expected_outputs=["daily_transactions_fees_by_to_v1"],
+    # TODO: Uncomment if we do this as a view (or some element as a view)
+    # auxiliary_views=[
+    #     TemplatedSQLQuery(
+    #         template_name="daily_transactions_fees_by_to",
+    #         context={},
+    #     ),
+    # ],
+)
+def daily_transactions_fees_by_to(duckdb_client: duckdb.DuckDBPyConnection) -> NamedRelations:
+    return {
+        "daily_transactions_fees_by_to_v1": duckdb_client.view(
+            """
+            TODO: AGGREGATION CODE
+            """
+        ),
+    }

src/op_analytics/datapipeline/models/code/event_emitting_transactions.py

@@ -0,0 +1,31 @@
+# # TO DEPRECATE?
+
+# import duckdb
+
+# from op_analytics.datapipeline.models.compute.querybuilder import TemplatedSQLQuery
+# from op_analytics.datapipeline.models.compute.registry import register_model
+# from op_analytics.datapipeline.models.compute.types import NamedRelations
+
+
+# @register_model(
+#     input_datasets=["ingestion/logs_v1", "ingestion/transactions_v1", "ingestion/blocks_v1"],
+#     expected_outputs=["event_emitting_transactions_v1"],
+#     auxiliary_views=[
+#         TemplatedSQLQuery(
+#             template_name="refined_transactions_fees",
+#             context={},
+#         ),
+#         TemplatedSQLQuery(
+#             template_name="logs_topic0_filters",
+#             context={},
+#         ),
+#         TemplatedSQLQuery(
+#             template_name="event_emitting_transactions",
+#             context={},
+#         ),
+#     ],
+# )
+# def event_emitting_transactions(duckdb_client: duckdb.DuckDBPyConnection) -> NamedRelations:
+#     return {
+#         "event_emitting_transactions_v1": duckdb_client.view("event_emitting_transactions"),
+#     }

src/op_analytics/datapipeline/models/code/refined_trace_calls.py

@@ -0,0 +1,21 @@
+# import duckdb
+
+# from op_analytics.datapipeline.models.compute.querybuilder import TemplatedSQLQuery
+# from op_analytics.datapipeline.models.compute.registry import register_model
+# from op_analytics.datapipeline.models.compute.types import NamedRelations
+
+
+# @register_model(
+#     input_datasets=["ingestion/traces_v1", "refined_transactions_fees_v1"],
+#     expected_outputs=["refined_trace_calls_v1"],
+#     auxiliary_views=[
+#         TemplatedSQLQuery(
+#             template_name="refined_trace_calls",
+#             context={},
+#         ),
+#     ],
+# )
+# def refined_trace_calls(duckdb_client: duckdb.DuckDBPyConnection) -> NamedRelations:
+#     return {
+#         "refined_trace_calls_v1": duckdb_client.view("refined_trace_calls"),
+#     }

src/op_analytics/datapipeline/models/code/refined_traces.py

@@ -1,5 +1,5 @@
-from op_analytics.coreutils.duckdb_inmem.client import DuckDBContext
-from op_analytics.datapipeline.models.compute.model import AuxiliaryView, ParquetData
+from op_analytics.coreutils.duckdb_inmem.client import DuckDBContext, ParquetData
+from op_analytics.datapipeline.models.compute.model import AuxiliaryView
 from op_analytics.datapipeline.models.compute.registry import register_model
 from op_analytics.datapipeline.models.compute.types import NamedRelations
 
@@ -10,18 +10,63 @@
         "ingestion/traces_v1",
         "ingestion/transactions_v1",
     ],
-    expected_outputs=[
-        "create_traces_v1",
-    ],
     auxiliary_views=[
         "refined_transactions_fees",
-        "refined_traces_fees",
+        "refined_traces/traces_projection",
+        "refined_traces/traces_amortized",
+        "refined_traces/traces_txs_join",
+    ],
+    expected_outputs=[
+        "refined_transactions_fees_v1",
+        "refined_traces_fees_v1",
     ],
 )
 def refined_traces(
     ctx: DuckDBContext,
     input_datasets: dict[str, ParquetData],
     auxiliary_views: dict[str, AuxiliaryView],
 ) -> NamedRelations:
-    # Not implemented yet.
-    return {}
+    refined_txs = auxiliary_views["refined_transactions_fees"].create_table(
+        duckdb_context=ctx,
+        template_parameters={
+            "raw_blocks": input_datasets["ingestion/blocks_v1"].as_subquery(),
+            "raw_transactions": input_datasets["ingestion/transactions_v1"].as_subquery(),
+        },
+    )
+
+    refined_traces_projection = auxiliary_views["refined_traces/traces_projection"].create_table(
+        duckdb_context=ctx,
+        template_parameters={
+            "raw_traces": input_datasets["ingestion/traces_v1"].as_subquery(),
+        },
+    )
+
+    traces_amortized = auxiliary_views["refined_traces/traces_amortized"].create_table(
+        duckdb_context=ctx,
+        template_parameters={
+            "refined_traces_projection": refined_traces_projection,
+        },
+    )
+
+    traces_txs_join = auxiliary_views["refined_traces/traces_txs_join"].create_table(
+        duckdb_context=ctx,
+        template_parameters={
+            "traces_amortized": traces_amortized,
+            "refined_transactions_fees": refined_txs,
+        },
+    )
+
+    ctx.client.sql(f"DROP TABLE {refined_traces_projection}")
+    ctx.client.sql(f"DROP TABLE {traces_amortized}")
+
+    return {
+        "refined_transactions_fees_v1": refined_txs,
+        "refined_traces_fees_v1": traces_txs_join,
+        # Also included in Michael's original implementation.
+        # "aggregated_traces_tr_from_tr_to_hash_v1": duckdb_client.view(
+        #     "aggregated_traces_tr_from_tr_to_hash"
+        # ),
+        # "aggregated_traces_tr_to_hash_v1": duckdb_client.view("aggregated_traces_tr_to_hash"),
+        # "daily_traces_tr_to_tx_to_v1": duckdb_client.view("daily_traces_tr_to_tx_to"),
+        # "daily_traces_tr_to_v1": duckdb_client.view("daily_traces_tr_to"),
+    }

src/op_analytics/datapipeline/models/code/refined_traces_models.py

@@ -0,0 +1,58 @@
+import duckdb
+
+from op_analytics.datapipeline.models.compute.querybuilder import TemplatedSQLQuery
+from op_analytics.datapipeline.models.compute.registry import register_model
+from op_analytics.datapipeline.models.compute.types import NamedRelations
+
+
+@register_model(
+    input_datasets=[
+        "ingestion/transactions_v1",
+        "ingestion/blocks_v1",
+        "ingestion/traces_v1",
+    ],
+    expected_outputs=[
+        "refined_traces_fees_v1",
+        "refined_traces_fees_tr_from_tr_to_hash_v1",
+        "daily_traces_tr_to_tx_to_v1",
+        "daily_traces_to_v1",
+    ],
+    auxiliary_views=[
+        TemplatedSQLQuery(
+            template_name="refined_transactions_fees",
+            context={},
+        ),
+        TemplatedSQLQuery(
+            template_name="refined_traces_fees",
+            context={},
+        ),
+        TemplatedSQLQuery(
+            template_name="aggregated_traces_tr_from_tr_to_hash",
+            context={},
+        ),
+        TemplatedSQLQuery(
+            template_name="aggregated_traces_tr_to_hash",
+            context={},
+        ),
+        TemplatedSQLQuery(
+            template_name="daily_traces_tr_to_tx_to",
+            context={},
+        ),
+        TemplatedSQLQuery(
+            template_name="daily_traces_tr_to",
+            context={},
+        ),
+    ],
+)
+def refined_traces_models(
+    duckdb_client: duckdb.DuckDBPyConnection,
+) -> NamedRelations:
+    return {
+        "refined_traces_fees_v1": duckdb_client.view("refined_traces_fees"),
+        "aggregated_traces_tr_from_tr_to_hash_v1": duckdb_client.view(
+            "aggregated_traces_tr_from_tr_to_hash"
+        ),
+        "aggregated_traces_tr_to_hash_v1": duckdb_client.view("aggregated_traces_tr_to_hash"),
+        "daily_traces_tr_to_tx_to_v1": duckdb_client.view("daily_traces_tr_to_tx_to"),
+        "daily_traces_tr_to_v1": duckdb_client.view("daily_traces_tr_to"),
+    }

src/op_analytics/datapipeline/models/code/refined_transactions_addresses_models.py

@@ -0,0 +1,65 @@
+import duckdb
+
+from op_analytics.datapipeline.models.compute.querybuilder import TemplatedSQLQuery
+from op_analytics.datapipeline.models.compute.registry import register_model
+from op_analytics.datapipeline.models.compute.types import NamedRelations
+
+
+@register_model(
+    input_datasets=[
+        "ingestion/transactions_v1",
+        "ingestion/blocks_v1",
+        "ingestion/logs_v1",
+    ],
+    expected_outputs=[
+        "refined_transactions_fees_v1",
+        "event_emitting_transactions_v1",
+        "summary_v1",
+        "daily_transactions_tx_from_tx_to_method_v1",
+        "daily_transactions_tx_to_method_v1",
+        "daily_transactions_tx_to_v1",
+    ],
+    auxiliary_views=[
+        TemplatedSQLQuery(
+            template_name="refined_transactions_fees",
+            context={},
+        ),
+        TemplatedSQLQuery(
+            template_name="logs_topic0_filters",
+            context={},
+        ),
+        TemplatedSQLQuery(
+            template_name="event_emitting_transactions",
+            context={},
+        ),
+        TemplatedSQLQuery(
+            template_name="daily_address_summary",
+            context={},
+        ),
+        TemplatedSQLQuery(
+            template_name="daily_transactions_tx_from_tx_to_method",
+            context={},
+        ),
+        TemplatedSQLQuery(
+            template_name="daily_transactions_tx_to_method",
+            context={},
+        ),
+        TemplatedSQLQuery(
+            template_name="daily_transactions_tx_to",
+            context={},
+        ),
+    ],
+)
+def refined_transactions_addresses_models(
+    duckdb_client: duckdb.DuckDBPyConnection,
+) -> NamedRelations:
+    return {
+        "refined_transactions_fees_v1": duckdb_client.view("refined_transactions_fees"),
+        "event_emitting_transactions_v1": duckdb_client.view("event_emitting_transactions"),
+        "summary_v1": duckdb_client.view("daily_address_summary"),
+        "daily_transactions_tx_from_tx_to_method_v1": duckdb_client.view(
+            "daily_transactions_tx_from_tx_to_method"
+        ),
+        "daily_transactions_tx_to_method_v1": duckdb_client.view("daily_transactions_tx_to_method"),
+        "daily_transactions_tx_to_v1": duckdb_client.view("daily_transactions_tx_to"),
+    }