AN-4280/heal-logic (#148)

* vars, readme, tx_success

* defi heal logic

* lookbacks on lps

* var format

* heal logic

* dex swaps heal

* pool name

* column names

README.md

@@ -35,35 +35,43 @@ gnosis:
       query_tag: <TAG>
 ```
 
-### Variables
+## Variables
 
 To control the creation of UDF or SP macros with dbt run:
 * UPDATE_UDFS_AND_SPS
-When True, executes all macros included in the on-run-start hooks within dbt_project.yml on model run as normal
-When False, none of the on-run-start macros are executed on model run
+  * Default values are False
+  * When True, executes all macros included in the on-run-start hooks within dbt_project.yml on model run as normal
+  * When False, none of the on-run-start macros are executed on model run
 
-Default values are False
+  * Usage: `dbt run --vars '{"UPDATE_UDFS_AND_SPS":True}' -m ...`
 
-* Usage:
-dbt run --var '{"UPDATE_UDFS_AND_SPS":True}'  -m ...
+Use a variable to heal a model incrementally:
+* HEAL_MODEL
+  * Default is FALSE (Boolean)
+  * When FALSE, logic will be negated
+  * When TRUE, heal logic will apply
+  * Include `heal` in model tags within the config block for inclusion in the `dbt_run_heal_models` workflow, e.g. `tags = 'heal'`
 
-To reload records in a curated complete table without a full-refresh, such as `silver_bridge.complete_bridge_activity`:
-* HEAL_CURATED_MODEL
-Default is an empty array []
-When item is included in var array [], incremental logic will be skipped for that CTE / code block  
-When item is not included in var array [] or does not match specified item in model, incremental logic will apply
-Example set up: `{% if is_incremental() and 'axelar' not in var('HEAL_CURATED_MODEL') %}`
+  * Usage: `dbt run --vars '{"HEAL_MODEL":True}' -m ...`
 
-* Usage:
-Single CTE: dbt run --var '{"HEAL_CURATED_MODEL":"axelar"}' -m ...
-Multiple CTEs: dbt run --var '{"HEAL_CURATED_MODEL":["axelar","across","celer_cbridge"]}' -m ...
+Use a variable to negate incremental logic:
+* Example use case: reload records in a curated complete table without a full-refresh, such as `silver_bridge.complete_bridge_activity`:
+* HEAL_MODELS
+  * Default is an empty array []
+  * When item is included in var array [], incremental logic will be skipped for that CTE / code block  
+  * When item is not included in var array [] or does not match specified item in model, incremental logic will apply
+  * Example set up: `{% if is_incremental() and 'axelar' not in var('HEAL_MODELS') %}`
 
-### Resources:
-- Learn more about dbt [in the docs](https://docs.getdbt.com/docs/introduction)
-- Check out [Discourse](https://discourse.getdbt.com/) for commonly asked questions and answers
-- Join the [chat](https://community.getdbt.com/) on Slack for live discussions and support
-- Find [dbt events](https://events.getdbt.com) near you
-- Check out [the blog](https://blog.getdbt.com/) for the latest news on dbt's development and best practices
+  * Usage:
+    * Single CTE: `dbt run --vars '{"HEAL_MODELS":"axelar"}' -m ...`
+    * Multiple CTEs: `dbt run --vars '{"HEAL_MODELS":["axelar","across","celer_cbridge"]}' -m ...`
+
+Use a variable to extend the incremental lookback period:
+* LOOKBACK
+  * Default is a string representing the specified time interval e.g. '12 hours', '7 days' etc.
+  * Example set up: `SELECT MAX(_inserted_timestamp) - INTERVAL '{{ var("LOOKBACK", "4 hours") }}'`
+
+  * Usage: `dbt run --vars '{"LOOKBACK":"36 hours"}' -m ...`
 
 ## Applying Model Tags
 
@@ -97,12 +105,19 @@ To add/update a model's snowflake tags, add/modify the `meta` model property und
 By default, model tags are pushed to Snowflake on each load. You can disable this by setting the `UPDATE_SNOWFLAKE_TAGS` project variable to `False` during a run.
 
 ```
-dbt run --var '{"UPDATE_SNOWFLAKE_TAGS":False}' -s models/core/core__fact_blocks.sql
+dbt run --vars '{"UPDATE_SNOWFLAKE_TAGS":False}' -s models/core/core__fact_blocks.sql
 ```
 
 ### Querying for existing tags on a model in snowflake
 
 ```
 select *
 from table(gnosis.information_schema.tag_references('gnosis.core.fact_blocks', 'table'));
-```
+```
+
+### Resources:
+- Learn more about dbt [in the docs](https://docs.getdbt.com/docs/introduction)
+- Check out [Discourse](https://discourse.getdbt.com/) for commonly asked questions and answers
+- Join the [chat](https://community.getdbt.com/) on Slack for live discussions and support
+- Find [dbt events](https://events.getdbt.com) near you
+- Check out [the blog](https://blog.getdbt.com/) for the latest news on dbt's development and best practices

dbt_project.yml

@@ -66,7 +66,7 @@ vars:
   WAIT: 0
   OBSERV_FULL_TEST: False
   HEAL_MODEL: False
-  HEAL_CURATED_MODEL: []
+  HEAL_MODELS: []
   START_GHA_TASKS: False
   API_INTEGRATION: '{{ var("config")[target.name]["API_INTEGRATION"] if var("config")[target.name] else var("config")["dev"]["API_INTEGRATION"] }}' 
   EXTERNAL_FUNCTION_URI: '{{ var("config")[target.name]["EXTERNAL_FUNCTION_URI"] if var("config")[target.name] else var("config")["dev"]["EXTERNAL_FUNCTION_URI"] }}'

models/gold/defi/defi__ez_bridge_activity.sql

@@ -33,7 +33,13 @@ SELECT
     token_symbol,
     amount_unadj,
     amount,
-    amount_usd,
+    ROUND(
+        CASE
+            WHEN amount_usd < 1e+15 THEN amount_usd
+            ELSE NULL
+        END,
+        2
+    ) AS amount_usd,
     COALESCE (
         complete_bridge_activity_id,
         {{ dbt_utils.generate_surrogate_key(

models/gold/defi/defi__ez_dex_swaps.sql

@@ -24,10 +24,26 @@ SELECT
   event_name,
   amount_in_unadj,
   amount_in,
-  amount_in_usd,
+  ROUND(
+        CASE
+            WHEN amount_out_usd IS NULL
+            OR ABS((amount_in_usd - amount_out_usd) / NULLIF(amount_out_usd, 0)) > 0.75
+            OR ABS((amount_in_usd - amount_out_usd) / NULLIF(amount_in_usd, 0)) > 0.75 THEN NULL
+            ELSE amount_in_usd
+        END,
+        2
+    ) AS amount_in_usd,
   amount_out_unadj,
   amount_out,
-  amount_out_usd,
+  ROUND(
+        CASE
+            WHEN amount_in_usd IS NULL
+            OR ABS((amount_out_usd - amount_in_usd) / NULLIF(amount_in_usd, 0)) > 0.75
+            OR ABS((amount_out_usd - amount_in_usd) / NULLIF(amount_out_usd, 0)) > 0.75 THEN NULL
+            ELSE amount_out_usd
+        END,
+        2
+    ) AS amount_out_usd,
   sender,
   tx_to,
   event_index,

models/silver/core/silver__transfers.sql

@@ -35,7 +35,7 @@ AND _inserted_timestamp >= (
     SELECT
         MAX(
             _inserted_timestamp
-        ) - INTERVAL '36 hours'
+        ) - INTERVAL '{{ var('LOOKBACK', '4 hours') }}'
     FROM
         {{ this }}
 )
@@ -169,7 +169,7 @@ heal_model AS (
                     SELECT
                         MAX(
                             _inserted_timestamp
-                        ) - INTERVAL '36 hours'
+                        ) - INTERVAL '{{ var('LOOKBACK', '4 hours') }}'
                     FROM
                         {{ this }}
                 )
@@ -195,7 +195,7 @@ heal_model AS (
                             SELECT
                                 MAX(
                                     _inserted_timestamp
-                                ) - INTERVAL '36 hours'
+                                ) - INTERVAL '{{ var('LOOKBACK', '4 hours') }}'
                             FROM
                                 {{ this }}
                         )

models/silver/defi/bridge/celer/silver_bridge__celer_cbridge_send.sql

@@ -46,6 +46,7 @@ WITH base_evt AS (
     WHERE
         topics [0] :: STRING = '0x89d8051e597ab4178a863a5190407b98abfeff406aa8db90c59af76612e58f01'
         AND contract_address = '0x3795c36e7d12a8c252a20c5a7b455f7c57b60283'
+        AND tx_status = 'SUCCESS'
 
 {% if is_incremental() %}
 AND _inserted_timestamp >= (

models/silver/defi/bridge/hop/silver_bridge__hop_ammwrapper.sql

@@ -14,6 +14,7 @@ WITH base_contracts AS (
         {{ ref('silver__logs') }}
     WHERE
         topics [0] :: STRING = '0xe35dddd4ea75d7e9b3fe93af4f4e40e778c3da4074c9d93e7c6536f1e803c1eb'
+        AND tx_status = 'SUCCESS'
 
 {% if is_incremental() %}
 AND _inserted_timestamp >= (

models/silver/defi/bridge/hop/silver_bridge__hop_transfersent.sql

@@ -51,6 +51,7 @@ WITH base_evt AS (
     WHERE
         topics [0] :: STRING = '0xe35dddd4ea75d7e9b3fe93af4f4e40e778c3da4074c9d93e7c6536f1e803c1eb'
         AND origin_to_address IS NOT NULL
+        AND tx_status = 'SUCCESS'
 
 {% if is_incremental() %}
 AND _inserted_timestamp >= (

models/silver/defi/bridge/meson/silver_bridge__meson_transfers.sql

@@ -126,6 +126,7 @@ dst_info AS (
     WHERE
         contract_address = '0x25ab3efd52e6470681ce037cd546dc60726948d3'
         AND topics [0] :: STRING = '0x5ce4019f772fda6cb703b26bce3ec3006eb36b73f1d3a0eb441213317d9f5e9d'
+        AND tx_status = 'SUCCESS'
 
 {% if is_incremental() %}
 AND _inserted_timestamp >= (