[examples] update examples for exchange shill calc (#7)

* sample cql files

* refactor ordertype

* shill refactor

* patch cypher query

---------

Co-authored-by: Montague McMarten <[email protected]>

src/analytics/enrich_account_funding.rs

@@ -9,7 +9,10 @@ use std::{
     io::Read,
 };
 
-use crate::schema_exchange_orders::ExchangeOrder;
+use crate::schema_exchange_orders::{ExchangeOrder, OrderType};
+
+#[cfg(test)]
+use crate::date_util::parse_date;
 
 #[cfg(test)]
 use crate::date_util::parse_date;
@@ -46,25 +49,22 @@ impl BalanceTracker {
 
     pub fn process_transaction_alt(&mut self, order: &ExchangeOrder) {
         let date = order.filled_at;
-        match order.order_type.as_str() {
-            "Buy" => {
+        match order.order_type {
+            OrderType::Buy => {
                 // user offered to buy coins (Buyer)
                 // he sends USD
                 // accepter sends coins. (Seller)
 
                 self.update_balance_and_flows_alt(order.user, date, order.amount, true);
                 self.update_balance_and_flows_alt(order.accepter, date, order.amount, false);
             }
-            "Sell" => {
+            OrderType::Sell => {
                 // user offered to sell coins (Seller)
                 // he sends Coins
                 // accepter sends USD. (Buyer)
                 self.update_balance_and_flows_alt(order.accepter, date, order.amount, true);
                 self.update_balance_and_flows_alt(order.user, date, order.amount, false);
             }
-            _ => {
-                println!("ERROR: not a valid Buy/Sell order, {:?}", &order);
-            }
         }
     }
     fn update_balance_and_flows_alt(
@@ -256,7 +256,7 @@ fn test_replay_transactions() {
         // user_1 sends USD, user_2 moves 10 coins.
         ExchangeOrder {
             user: 1,
-            order_type: "Buy".to_string(),
+            order_type: OrderType::Buy,
             amount: 10.0,
             price: 2.0,
             created_at: parse_date("2024-03-01"),
@@ -266,13 +266,13 @@ fn test_replay_transactions() {
             rms_24hour: 0.0,
             price_vs_rms_hour: 0.0,
             price_vs_rms_24hour: 0.0,
-            shill_bid: None,
+            ..Default::default()
         },
         ExchangeOrder {
             // user 2 creates an offer to SELL, user 3 accepts.
             // user 3 sends USD user 2 moves amount of coins.
             user: 2,
-            order_type: "Sell".to_string(),
+            order_type: OrderType::Sell,
             amount: 5.0,
             price: 3.0,
             created_at: parse_date("2024-03-05"),
@@ -282,13 +282,13 @@ fn test_replay_transactions() {
             rms_24hour: 0.0,
             price_vs_rms_hour: 0.0,
             price_vs_rms_24hour: 0.0,
-            shill_bid: None,
+            ..Default::default()
         },
         // user 3 creates an offer to BUY, user 1 accepts.
         // user 3 sends USD user 1 moves amount of coins.
         ExchangeOrder {
             user: 3,
-            order_type: "Buy".to_string(),
+            order_type: OrderType::Buy,
             amount: 15.0,
             price: 1.5,
             created_at: parse_date("2024-03-10"),
@@ -298,7 +298,7 @@ fn test_replay_transactions() {
             rms_24hour: 0.0,
             price_vs_rms_hour: 0.0,
             price_vs_rms_24hour: 0.0,
-            shill_bid: None,
+            ..Default::default()
         },
     ];
 

src/analytics/enrich_rms.rs

@@ -1,7 +1,7 @@
 use chrono::Duration;
 use std::collections::VecDeque;
 
-use crate::schema_exchange_orders::ExchangeOrder;
+use crate::schema_exchange_orders::{CompetingOffers, ExchangeOrder, OrderType};
 
 fn calculate_rms(data: &[f64]) -> f64 {
     let (sum, count) = data
@@ -80,93 +80,75 @@ pub fn include_rms_stats(swaps: &mut [ExchangeOrder]) {
     }
 }
 
-pub fn process_sell_order_shill(swaps: &mut [ExchangeOrder]) {
-    swaps.sort_by_key(|swap| swap.filled_at); // Sort by filled_at
-
-    for i in 0..swaps.len() {
-        let current_swap = &swaps[i];
-        // TODO: move this to a filter on the enclosing scope
-        if current_swap.shill_bid.is_some() {
+fn get_competing_offers(
+    current_order: &ExchangeOrder,
+    all_offers: &[ExchangeOrder],
+) -> CompetingOffers {
+    let mut competition = CompetingOffers {
+        offer_type: current_order.order_type.clone(),
+        ..Default::default()
+    };
+
+    for other in all_offers {
+        if competition.offer_type != other.order_type {
             continue;
-        };
-
-        // Filter for open trades
-        let open_orders = swaps
-            .iter()
-            .filter(|&other_swap| {
-                other_swap.filled_at > current_swap.filled_at
-                    && other_swap.created_at <= current_swap.filled_at
-            })
-            .collect::<Vec<_>>();
+        }
 
-        // Determine if the current swap took the best price
-        let is_shill_bid = match current_swap.order_type.as_str() {
-            // Signs of shill trades.
-            // For those offering to SELL coins, as the tx.user (offerer)
-            // I should offer to sell near the current clearing price.
-            // If I'm making shill bids, I'm creating trades above the current clearing price. An honest actor wouldn't expect those to get filled immediately.
-            // If an accepter is buying coins at a higher price than other orders which could be filled, then they are likely colluding to increase the price.
-            "Sell" => open_orders.iter().any(|other_swap|
-                  // if there are cheaper SELL offers,
-                  // for smaller sizes, then the rational honest actor
-                  // will pick one of those.
-                  // So we find the list of open orders which would be
-                  // better than the one taken how.
-                  // if there are ANY available, then this SELL order was
-                  // filled dishonestly.
-                  other_swap.price <= current_swap.price &&
-                  other_swap.amount <= current_swap.amount),
-            _ => false,
-        };
+        // is the other offer created in the past, and still not filled
+        if other.created_at < current_order.filled_at && other.filled_at > current_order.filled_at {
+            competition.open_same_type += 1;
+            if other.amount <= current_order.amount {
+                competition.within_amount += 1;
 
-        // Update the swap with the best price flag
-        swaps[i].shill_bid = Some(is_shill_bid);
+                if other.price <= current_order.price {
+                    competition.within_amount_lower_price += 1;
+                }
+            }
+        }
     }
-}
-
-pub fn process_buy_order_shill(swaps: &mut [ExchangeOrder]) {
-    // NEED to sort by created_at to identify shill created BUY orders
-    swaps.sort_by_key(|swap| swap.created_at);
-
-    for i in 0..swaps.len() {
-        let current_swap = &swaps[i];
-
-        // TODO: move this to a filter on the enclosing scope
-        if current_swap.shill_bid.is_some() {
-            continue;
-        };
-
-        // Filter for open trades
-        let open_orders = swaps
-            .iter()
-            .filter(|&other_swap| {
-                other_swap.filled_at > current_swap.created_at
-                    && other_swap.created_at <= current_swap.created_at
-            })
-            .collect::<Vec<_>>();
 
-        // Determine if the current swap took the best price
-        let is_shill_bid = match current_swap.order_type.as_str() {
-            // Signs of shill trades.
-            // For those offering to BUY coins, as the tx.user (offerer)
-            // An honest and rational actor would not create a buy order
-            // higher than other SELL offers which have not been filled.
-            // The shill bidder who is colluding will create a BUY order at a higher price than other SELL orders which currently exist.
-            "Buy" => open_orders.iter().any(|other_swap| {
-                if other_swap.order_type == *"Sell" {
-                    // this is not a rational trade if there are
-                    // SELL offers of the same amount (or smaller)
-                    // at a price equal or lower.
-                    return other_swap.price <= current_swap.price
-                        && other_swap.amount <= current_swap.amount;
+    competition
+}
+pub fn process_shill(all_transactions: &mut [ExchangeOrder]) {
+    all_transactions.sort_by_key(|el| el.filled_at); // Sort by filled_at
+
+    // TODO: gross, see what you make me do, borrow checker.
+    let temp_tx = all_transactions.to_vec();
+
+    for current_order in all_transactions.iter_mut() {
+        let comp = get_competing_offers(current_order, &temp_tx);
+
+        // We can only evaluate if an "accepter" is engaged in shill behavior.
+        // the "offerer" may create unreasonable offers, but the shill trade requires someone accepting.
+
+        match comp.offer_type {
+            // An accepter may be looking to dispose of coins.
+            // They must fill someone else's "BUY" offer.
+
+            // Rationally would want to dispose at the highest price possible.
+            // so if we find that there were more HIGHER offers to buy which this accepter did not take, we must wonder why they are taking a lower price voluntarily.
+            // it would indicate they are shilling_down
+            OrderType::Buy => {
+                if let Some(higher_priced_orders) = comp
+                    .within_amount
+                    .checked_sub(comp.within_amount_lower_price)
+                {
+                    if higher_priced_orders > 0 {
+                        current_order.accepter_shill_down = true
+                    }
                 }
-                false
-            }),
-            _ => false,
-        };
-
-        // Update the swap with the best price flag
-        swaps[i].shill_bid = Some(is_shill_bid);
+                // Similarly an accepter may be looking to accumulate coins.
+                // They rationally will do so at the lowest price available
+                // We want to check if they are ignoring lower priced offers
+                // of the same or lower amount.
+                // If so it means they are pushing the price up.
+            }
+            OrderType::Sell => {
+                if comp.within_amount_lower_price > 0 {
+                    current_order.accepter_shill_up = true
+                }
+            }
+        }
     }
 }
 
@@ -186,12 +168,12 @@ fn test_rms_pipeline() {
                 .unwrap()
                 .with_timezone(&Utc),
             price: 100.0,
-            order_type: "Buy".into(),
+            order_type: OrderType::Buy,
             rms_hour: 0.0,
             rms_24hour: 0.0,
             price_vs_rms_hour: 0.0,
             price_vs_rms_24hour: 0.0,
-            shill_bid: None,
+            ..Default::default()
         },
         // less than 12 hours later next trade 5/6/2024 8AM
         ExchangeOrder {
@@ -205,12 +187,12 @@ fn test_rms_pipeline() {
                 .unwrap()
                 .with_timezone(&Utc),
             price: 4.0,
-            order_type: "Buy".into(),
+            order_type: OrderType::Buy,
             rms_hour: 0.0,
             rms_24hour: 0.0,
             price_vs_rms_hour: 0.0,
             price_vs_rms_24hour: 0.0,
-            shill_bid: None,
+            ..Default::default()
         },
         // less than one hour later
         ExchangeOrder {
@@ -224,12 +206,12 @@ fn test_rms_pipeline() {
                 .unwrap()
                 .with_timezone(&Utc),
             price: 4.0,
-            order_type: "Buy".into(),
+            order_type: OrderType::Buy,
             rms_hour: 0.0,
             rms_24hour: 0.0,
             price_vs_rms_hour: 0.0,
             price_vs_rms_24hour: 0.0,
-            shill_bid: None,
+            ..Default::default()
         },
         // same time as previous but different traders
         ExchangeOrder {
@@ -243,12 +225,7 @@ fn test_rms_pipeline() {
                 .unwrap()
                 .with_timezone(&Utc),
             price: 32.0,
-            order_type: "Sell".into(),
-            rms_hour: 0.0,
-            rms_24hour: 0.0,
-            price_vs_rms_hour: 0.0,
-            price_vs_rms_24hour: 0.0,
-            shill_bid: None,
+            ..Default::default()
         },
     ];
 
@@ -267,6 +244,6 @@ fn test_rms_pipeline() {
     assert!(s3.rms_hour == 4.0);
     assert!((s3.rms_24hour > 57.0) && (s3.rms_24hour < 58.0));
 
-    process_sell_order_shill(&mut swaps);
+    process_shill(&mut swaps);
     dbg!(&swaps);
 }

src/load_exchange_orders.rs

@@ -10,7 +10,7 @@ use crate::{
     schema_exchange_orders::ExchangeOrder,
 };
 
-pub async fn swap_batch(
+pub async fn exchange_txs_batch(
     txs: &[ExchangeOrder],
     pool: &Graph,
     batch_size: usize,
@@ -93,14 +93,13 @@ pub async fn load_from_json(path: &Path, pool: &Graph, batch_size: usize) -> Res
     info!("completed rms statistics");
 
     // find likely shill bids
-    enrich_rms::process_sell_order_shill(&mut orders);
-    enrich_rms::process_buy_order_shill(&mut orders);
+    enrich_rms::process_shill(&mut orders);
     info!("completed shill bid calculation");
 
     let mut balances = BalanceTracker::new();
     balances.replay_transactions(&mut orders)?;
     let ledger_inserts = balances.submit_ledger(pool).await?;
-    info!("exchange ledger relations inserted: {}", ledger_inserts);
+    info!("exchange UserLedger state inserted: {}", ledger_inserts);
 
-    swap_batch(&orders, pool, batch_size).await
+    exchange_txs_batch(&orders, pool, batch_size).await
 }

src/neo4j_init.rs

@@ -29,6 +29,9 @@ pub static INDEX_TX_HASH: &str =
 pub static INDEX_TX_FUNCTION: &str =
     "CREATE INDEX tx_function IF NOT EXISTS FOR ()-[r:Tx]-() ON (r.function)";
 
+pub static INDEX_TX_RELATION: &str =
+    "CREATE INDEX tx_relation IF NOT EXISTS FOR ()-[r:Tx]-() ON (r.relation)";
+
 pub static INDEX_SWAP_ID: &str =
     "CREATE INDEX swap_account_id IF NOT EXISTS FOR (n:SwapAccount) ON (n.swap_id)";
 
@@ -74,6 +77,7 @@ pub async fn maybe_create_indexes(graph: &Graph) -> Result<()> {
         INDEX_TX_TIMESTAMP,
         INDEX_TX_HASH,
         INDEX_TX_FUNCTION,
+        INDEX_TX_RELATION,
         INDEX_SWAP_ID,
         INDEX_EXCHANGE_LEDGER,
         INDEX_EXCHANGE_LINK_LEDGER,