better tracking of type 2 changes

Makefile

@@ -32,5 +32,5 @@ run: build
 
 .PHONY: gui
 gui: build
-	substreams gui substreams.yaml map_balance_changes -e mainnet.eth.streamingfast.io:443 -s 17000000 -t +1000 --production-mode
+	substreams gui substreams.yaml balance_change_stats -e mainnet.eth.streamingfast.io:443 --production-mode
 

README.md

@@ -88,11 +88,11 @@ These types of transfers will result in a BalanceChange message with `change_typ
 
 ### Type 2: Storage change is in a different call than the transfer
 
+In this case, the Transfer but this results in storage changes in different child calls, where often the amount sent will be split to multiple accounts.
+
 example:
 https://etherscan.io/tx/0x5a31fb5d3f5bbb95023438f017ad6cd501ce70e445f31c2660c784e5a7eb5d83#eventlog
 
-In this case, the Transfer is done in call index 4, but the storage change is actually recorded in call index 10.  Here is the relevant section from the Firehose block for this transaction:
-
 ```json
 {
   "index": 4,
@@ -128,9 +128,7 @@ In this case, the Transfer is done in call index 4, but the storage change is ac
 }
 ```
 
-So some logic is required to track the storage change to the transfer.
-
-The correctness of the `old_balance` and `new_balance` values in this case is not as easily determined.  It is an open question at this point as to whether the values given in the `oldValue` and `newValue` fields always correspond to the correct balance values.
+In this example, the Transfer call is made in call index 4.  Then in the subsequent child calls, the transfer of 4,530,000 tokens is split into transfers by the contract:  One transfer of 4,500,000 to the original receiver and a transfer of 30,000 to another address.  Some work is required to track the balance changes in this case.
 
 These types of transfers will result in a BalanceChange message with `change_type` set to `TYPE_2`.
 
@@ -143,4 +141,19 @@ https://etherscan.io/tx/0x5a31fb5d3f5bbb95023438f017ad6cd501ce70e445f31c2660c784
 
 These transfers will result in a BalanceChange message with `change_type` set to `null`.
 
-These should currently be discarded by the consumer of the substream as they are guaranteed to be incorrect.
+These should currently be discarded by the consumer of the substream as they are guaranteed to be incorrect.
+
+### Notes
+
+As of block 18005744, the sum of type 1 and type 2 matches accounts for approximately 96.7% of the total balance changes.
+
+```json
+{
+  "type0Count": "88809770",
+  "type1Count": "2608546600",
+  "type2Count": "5195308",
+  "totalCount": "2702551678",
+  "validRate": "0.9671385488303694890529305171717793142603506581308747872905614795055919001005685856860791544131205324",
+  "blockNumber": "18005744"
+}
+```

proto/v1/erc20.proto

@@ -28,8 +28,11 @@ message BalanceChange {
 }
 
 message BalanceChangeStats {
-  uint32 total_changes = 1;
-  uint32 type_1_changes = 2;
-  uint32 type_2_changes = 3;
-  uint32 unknown_changes = 4;
+  uint64 type0_count = 1;
+  uint64 type1_count = 2;
+  uint64 type2_count = 3;
+
+  uint64 total_count = 42;
+  string valid_rate = 43;
+  uint64 block_number = 99;
 }

src/lib.rs

@@ -1,4 +1,5 @@
 mod maps;
 mod sinks;
 mod abi;
-mod pb;
+mod pb;
+mod store;

src/maps.rs

@@ -1,12 +1,14 @@
 use crate::abi::{self};
-use crate::pb::erc20::types::v1::{BalanceChange, BalanceChangeType, BalanceChanges};
+use crate::pb::erc20::types::v1::{BalanceChange, BalanceChangeType, BalanceChanges, BalanceChangeStats};
 use abi::erc20::events::Transfer;
 use hex_literal::hex;
 use std::collections::HashMap;
 use substreams::errors::Error;
 use substreams::log::info;
-use substreams::scalar::BigInt;
+use substreams::scalar::{BigDecimal, BigInt};
 use substreams::Hex;
+use substreams::pb::substreams::Clock;
+use substreams::store::{StoreGet, StoreGetBigInt};
 use substreams_ethereum::pb::eth::v2::{Block, Call, TransactionTrace, TransactionTraceStatus};
 use substreams_ethereum::Event;
 
@@ -20,6 +22,28 @@ pub fn map_balance_changes(block: Block) -> Result<BalanceChanges, Error> {
     })
 }
 
+#[substreams::handlers::map]
+pub fn balance_change_stats(clock: Clock, store: StoreGetBigInt) -> Result<BalanceChangeStats, Error> {
+
+
+    let type_1 = store.get_last("type1").unwrap_or(BigInt::from(0));
+    let type_2 = store.get_last("type2").unwrap_or(BigInt::from(0));
+    let total = store.get_last("total").unwrap_or(BigInt::from(0));
+    let mut valid_rate = BigDecimal::from(0);
+    if !total.is_zero() {
+        valid_rate = (BigDecimal::from(type_1.clone()) + BigDecimal::from(type_2.clone())) / BigDecimal::from(total.clone());
+    }
+
+    Ok(BalanceChangeStats {
+        type0_count: store.get_last("type0").unwrap_or(BigInt::from(0)).to_u64(),
+        type1_count: type_1.to_u64(),
+        type2_count: type_2.to_u64(),
+        total_count: total.to_u64(),
+        block_number: clock.number,
+        valid_rate: valid_rate.to_string(),
+    })
+}
+
 pub fn map_balance_change(block: Block) -> Vec<BalanceChange> {
     let mut balance_changes = Vec::new();
 

src/pb/erc20.types.v1.rs

@@ -38,14 +38,18 @@ pub struct BalanceChange {
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct BalanceChangeStats {
-    #[prost(uint32, tag="1")]
-    pub total_changes: u32,
-    #[prost(uint32, tag="2")]
-    pub type_1_changes: u32,
-    #[prost(uint32, tag="3")]
-    pub type_2_changes: u32,
-    #[prost(uint32, tag="4")]
-    pub unknown_changes: u32,
+    #[prost(uint64, tag="1")]
+    pub type0_count: u64,
+    #[prost(uint64, tag="2")]
+    pub type1_count: u64,
+    #[prost(uint64, tag="3")]
+    pub type2_count: u64,
+    #[prost(uint64, tag="42")]
+    pub total_count: u64,
+    #[prost(string, tag="43")]
+    pub valid_rate: ::prost::alloc::string::String,
+    #[prost(uint64, tag="99")]
+    pub block_number: u64,
 }
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
 #[repr(i32)]

src/store.rs

@@ -0,0 +1,34 @@
+use substreams::scalar::BigInt;
+use substreams::store::{StoreAdd, StoreAddBigInt};
+use crate::pb::erc20::types::v1::{BalanceChanges};
+use substreams::store::StoreNew;
+
+#[substreams::handlers::store]
+pub fn store_valid_balance_changes(balance_changes: BalanceChanges, store: StoreAddBigInt) {
+    let mut type0_counter : u64 = 0;
+    let mut type1_counter : u64 = 0;
+    let mut type2_counter : u64 = 0;
+    let mut total_counter : u64 = 0;
+
+    for change in balance_changes.balance_changes {
+        match change.change_type {
+            1 => {
+                type1_counter += 1;
+                total_counter += 1;
+            },
+            2 => {
+                type2_counter += 1;
+                total_counter += 1;
+            },
+            _ => {
+                type0_counter += 1;
+                total_counter += 1;
+            },
+        }
+    }
+
+    store.add(0, "type1", BigInt::from(type1_counter));
+    store.add(0, "type2", BigInt::from(type2_counter));
+    store.add(0, "type0", BigInt::from(type0_counter));
+    store.add(0, "total", BigInt::from(total_counter));
+}

substreams.yaml

@@ -1,7 +1,7 @@
 specVersion: v0.1.0
 package:
   name: erc20_balance_changes
-  version: v0.0.2
+  version: v0.0.3
   url: https://github.com/streamingfast/erc20-balance-changes
   doc: ERC-20
 
@@ -46,3 +46,18 @@ modules:
       - map: map_balance_changes
     output:
       type: proto:sf.substreams.sink.entity.v1.EntityChanges
+
+  - name: store_valid_balance_changes
+    kind: store
+    updatePolicy: add
+    valueType: bigint
+    inputs:
+      - map: map_balance_changes
+
+  - name: balance_change_stats
+    kind: map
+    inputs:
+      - source: sf.substreams.v1.Clock
+      - store: store_valid_balance_changes
+    output:
+      type: proto:erc20.types.v1.BalanceChangeStats