Unify the code across the arb-mode classes

fastlane_bot/modes/pairwise_multi_all.py

@@ -20,23 +20,24 @@ def get_combos(self) -> List[Any]:
         return [(tkn0, tkn1) for tkn0, tkn1 in product(all_tokens, flashloan_tokens_intersect) if tkn0 != tkn1]
 
     def get_curve_combos(self, CC: Any) -> List[Any]:
-        carbon_curves = [x for x in CC.curves if x.params.exchange in self.ConfigObj.CARBON_V1_FORKS]
-        non_carbon_curves = [x for x in CC.curves if x.params.exchange not in self.ConfigObj.CARBON_V1_FORKS]
-        curve_combos = [[_curve0] + [_curve1] for _curve0 in non_carbon_curves for _curve1 in non_carbon_curves if _curve0 != _curve1]
+        carbon_curves = [curve for curve in CC.curves if curve.params.exchange in self.ConfigObj.CARBON_V1_FORKS]
+        other_curves = [curve for curve in CC.curves if curve.params.exchange not in self.ConfigObj.CARBON_V1_FORKS]
 
         if len(carbon_curves) > 0:
-            base_direction_pair = carbon_curves[0].pair
-            base_direction_one = [curve for curve in carbon_curves if curve.pair == base_direction_pair]
-            base_direction_two = [curve for curve in carbon_curves if curve.pair != base_direction_pair]
             curve_combos = []
 
-            if len(base_direction_one) > 0:
-                curve_combos += [[curve] + base_direction_one for curve in non_carbon_curves]
+            base_dir_one = [curve for curve in carbon_curves if curve.pair == carbon_curves[0].pair]
+            base_dir_two = [curve for curve in carbon_curves if curve.pair != carbon_curves[0].pair]
 
-            if len(base_direction_two) > 0:
-                curve_combos += [[curve] + base_direction_two for curve in non_carbon_curves]
+            if len(base_dir_one) > 0:
+                curve_combos += [[curve] + base_dir_one for curve in other_curves]
 
-            if len(carbon_curves) >= 2:
+            if len(base_dir_two) > 0:
+                curve_combos += [[curve] + base_dir_two for curve in other_curves]
+
+            if len(carbon_curves) > 1:
                 curve_combos += [carbon_curves]
 
-        return curve_combos
+            return curve_combos
+
+        return [[curve0, curve1] for curve0 in other_curves for curve1 in other_curves if curve0 != curve1]

fastlane_bot/modes/pairwise_multi_pol.py

@@ -20,20 +20,19 @@ def get_combos(self) -> List[Any]:
         return [(tkn0, tkn1) for tkn0, tkn1 in product(bancor_pol_tkns, [self.ConfigObj.WETH_ADDRESS]) if tkn0 != tkn1]
 
     def get_curve_combos(self, CC: Any) -> List[Any]:
-        pol_curves = [x for x in CC.curves if x.params.exchange == "bancor_pol"]
-        carbon_curves = [x for x in CC.curves if x.params.exchange in self.ConfigObj.CARBON_V1_FORKS]
-        non_carbon_curves = [x for x in CC.curves if x.params.exchange not in ["bancor_pol"] + self.ConfigObj.CARBON_V1_FORKS]
-        curve_combos = [[curve] + pol_curves for curve in non_carbon_curves]
+        pol_curves = [curve for curve in CC.curves if curve.params.exchange == "bancor_pol"]
+        carbon_curves = [curve for curve in CC.curves if curve.params.exchange in self.ConfigObj.CARBON_V1_FORKS]
+        other_curves = [curve for curve in CC.curves if curve.params.exchange not in ["bancor_pol"] + self.ConfigObj.CARBON_V1_FORKS]
+        curve_combos = [[curve] + pol_curves for curve in other_curves]
 
         if len(carbon_curves) > 0:
-            base_direction_pair = carbon_curves[0].pair
-            base_direction_one = [curve for curve in carbon_curves if curve.pair == base_direction_pair]
-            base_direction_two = [curve for curve in carbon_curves if curve.pair != base_direction_pair]
+            base_dir_one = [curve for curve in carbon_curves if curve.pair == carbon_curves[0].pair]
+            base_dir_two = [curve for curve in carbon_curves if curve.pair != carbon_curves[0].pair]
 
-            if len(base_direction_one) > 0:
-                curve_combos += [[curve] + base_direction_one for curve in pol_curves]
+            if len(base_dir_one) > 0:
+                curve_combos += [[curve] + base_dir_one for curve in pol_curves]
 
-            if len(base_direction_two) > 0:
-                curve_combos += [[curve] + base_direction_two for curve in pol_curves]
+            if len(base_dir_two) > 0:
+                curve_combos += [[curve] + base_dir_two for curve in pol_curves]
 
         return curve_combos

fastlane_bot/modes/triangle_bancor_v3_two_hop.py

@@ -24,37 +24,35 @@ def get_combos(self) -> List[Any]:
             flashloan_tokens = fltkns
 
         miniverse_combos = []
-        combos = [(tkn0, tkn1) for tkn0, tkn1 in product(flashloan_tokens, flashloan_tokens) if tkn0 != tkn1]
 
-        for tkn0, tkn1 in combos:
+        for tkn0, tkn1 in [(tkn0, tkn1) for tkn0, tkn1 in product(flashloan_tokens, flashloan_tokens) if tkn0 != tkn1]:
             all_curves = list(set(self.CCm.bypairs(f"{tkn0}/{tkn1}")) | set(self.CCm.bypairs(f"{tkn1}/{tkn0}")))
 
             carbon_curves = [curve for curve in all_curves if curve.params.get("exchange") in self.ConfigObj.CARBON_V1_FORKS]
-            if not carbon_curves:
+            if len(carbon_curves) == 0:
                 continue
 
             external_curves = [curve for curve in all_curves if curve.params.get("exchange") not in self.ConfigObj.CARBON_V1_FORKS]
-            if not external_curves:
+            if len(external_curves) == 0:
                 continue
 
             bancor_v3_curve_0 = self.CCm.bypairs(f"{self.ConfigObj.BNT_ADDRESS}/{tkn0}").byparams(exchange="bancor_v3").curves
-            if not bancor_v3_curve_0:
+            if len(bancor_v3_curve_0) == 0:
                 continue
 
             bancor_v3_curve_1 = self.CCm.bypairs(f"{self.ConfigObj.BNT_ADDRESS}/{tkn1}").byparams(exchange="bancor_v3").curves
-            if not bancor_v3_curve_1:
+            if len(bancor_v3_curve_1) == 0:
                 continue
 
+            base_dir_one = [curve for curve in carbon_curves if curve.pair == carbon_curves[0].pair]
+            base_dir_two = [curve for curve in carbon_curves if curve.pair != carbon_curves[0].pair]
             miniverses = [bancor_v3_curve_0 + bancor_v3_curve_1 + [curve] for curve in external_curves]
 
-            base_direction_one = [curve for curve in carbon_curves if curve.pair == carbon_curves[0].pair]
-            base_direction_two = [curve for curve in carbon_curves if curve.pair != carbon_curves[0].pair]
+            if len(base_dir_one) > 0:
+                miniverses += [bancor_v3_curve_0 + bancor_v3_curve_1 + base_dir_one]
 
-            if len(base_direction_one) > 0:
-                miniverses += [bancor_v3_curve_0 + bancor_v3_curve_1 + base_direction_one]
-
-            if len(base_direction_two) > 0:
-                miniverses += [bancor_v3_curve_0 + bancor_v3_curve_1 + base_direction_two]
+            if len(base_dir_two) > 0:
+                miniverses += [bancor_v3_curve_0 + bancor_v3_curve_1 + base_dir_two]
 
             miniverses += [bancor_v3_curve_0 + bancor_v3_curve_1 + carbon_curves]
 

fastlane_bot/modes/triangle_multi.py

@@ -16,65 +16,64 @@
 class ArbitrageFinderTriangleMulti(ArbitrageFinderTriangleBase):
     def get_combos(self) -> List[Any]:
         combos = []
-        all_base_exchange_curves = self.CCm.byparams(exchange="carbon_v1").curves
+        all_carbon_curves = self.CCm.byparams(exchange="carbon_v1").curves
 
-        for flt in self.flashloan_tokens:  # may wish to run this for one flt at a time
-            non_flt_base_exchange_curves = [x for x in all_base_exchange_curves if flt not in x.pair]
-            for non_flt_base_exchange_curve in non_flt_base_exchange_curves:
-                target_tkny = non_flt_base_exchange_curve.tkny
-                target_tknx = non_flt_base_exchange_curve.tknx
+        for flt in self.flashloan_tokens:
+            non_flt_carbon_curves = [curve for curve in all_carbon_curves if flt not in curve.pair]
+            for non_flt_carbon_curve in non_flt_carbon_curves:
+                tkny = non_flt_carbon_curve.tkny
+                tknx = non_flt_carbon_curve.tknx
 
-                base_exchange_curves = self.CCm.bypairs(f"{target_tknx}/{target_tkny}").byparams(exchange="carbon_v1").curves
-                if len(base_exchange_curves) == 0:
+                carbon_curves = self.CCm.bypairs(f"{tknx}/{tkny}").byparams(exchange="carbon_v1").curves
+                if len(carbon_curves) == 0:
                     continue
 
-                base_direction_pair = base_exchange_curves[0].pair
-                base_direction_one = [curve for curve in base_exchange_curves if curve.pair == base_direction_pair]
-                base_direction_two = [curve for curve in base_exchange_curves if curve.pair != base_direction_pair]
+                y_match_curves = self.CCm.bypairs(set(self.CCm.filter_pairs(onein=tknx)) & set(self.CCm.filter_pairs(onein=flt)))
+                x_match_curves = self.CCm.bypairs(set(self.CCm.filter_pairs(onein=tkny)) & set(self.CCm.filter_pairs(onein=flt)))
 
-                y_match_curves = self.CCm.bypairs(set(self.CCm.filter_pairs(onein=target_tknx)) & set(self.CCm.filter_pairs(onein=flt)))
-                x_match_curves = self.CCm.bypairs(set(self.CCm.filter_pairs(onein=target_tkny)) & set(self.CCm.filter_pairs(onein=flt)))
-
-                y_match_curves_not_carbon = [x for x in y_match_curves if x.params.exchange != "carbon_v1"]
-                if len(y_match_curves_not_carbon) == 0:
+                y_match_other_curves = [curve for curve in y_match_curves if curve.params.exchange != "carbon_v1"]
+                if len(y_match_other_curves) == 0:
                     continue
 
-                x_match_curves_not_carbon = [x for x in x_match_curves if x.params.exchange != "carbon_v1"]
-                if len(x_match_curves_not_carbon) == 0:
+                x_match_other_curves = [curve for curve in x_match_curves if curve.params.exchange != "carbon_v1"]
+                if len(x_match_other_curves) == 0:
                     continue
 
-                if len(base_direction_one) > 0:
-                    combos += get_miniverse_combos(y_match_curves_not_carbon, base_direction_one, x_match_curves_not_carbon, flt)
+                base_dir_one = [curve for curve in carbon_curves if curve.pair == carbon_curves[0].pair]
+                base_dir_two = [curve for curve in carbon_curves if curve.pair != carbon_curves[0].pair]
+
+                if len(base_dir_one) > 0:
+                    combos += get_miniverse_combos(y_match_other_curves, x_match_other_curves, base_dir_one, flt)
 
-                if len(base_direction_two) > 0:
-                    combos += get_miniverse_combos(y_match_curves_not_carbon, base_direction_two, x_match_curves_not_carbon, flt)
+                if len(base_dir_two) > 0:
+                    combos += get_miniverse_combos(y_match_other_curves, x_match_other_curves, base_dir_two, flt)
 
         return combos
 
 def get_miniverse_combos(
-    y_match_curves_not_carbon: List[Any],
+    y_match_other_curves: List[Any],
+    x_match_other_curves: List[Any],
     base_exchange_curves: List[Any],
-    x_match_curves_not_carbon: List[Any],
     flt: str
 ):
     """
     Get miniverse for triangular arbitrage
 
     Parameters
     ----------
-    y_match_curves_not_carbon : list
+    y_match_other_curves : list
         List of curves that match the y token and are not on carbon
-    base_exchange_curves : list
-        List of curves on the base exchange
-    x_match_curves_not_carbon : list
+    x_match_other_curves : list
         List of curves that match the x token and are not on carbon
+    base_exchange_curves : list
+        List of curves on carbon
     flt : str
         Flashloan token
 
     Returns
     -------
     A list of miniverse combos
     """
-    curve_combos = list(product(y_match_curves_not_carbon, x_match_curves_not_carbon))
+    curve_combos = list(product(y_match_other_curves, x_match_other_curves))
     miniverses = [base_exchange_curves + list(combo) for combo in curve_combos]
     return [(flt, miniverse) for miniverse in miniverses]

fastlane_bot/modes/triangle_multi_complete.py

@@ -19,17 +19,17 @@ def get_combos(self) -> List[Any]:
 
         for flt in self.flashloan_tokens:
             # Get the Carbon pairs
-            carbon_pairs = sort_pairs(set([x.pair for x in self.CCm.curves if x.params.exchange in self.ConfigObj.CARBON_V1_FORKS]))
-            
-            # Create a set of unique tokens, excluding 'flt'
-            x_tokens = {token for pair in carbon_pairs for token in pair.split('/') if token != flt}
-            
+            carbon_pairs = sort_pairs(set([curve.pair for curve in self.CCm.curves if curve.params.exchange in self.ConfigObj.CARBON_V1_FORKS]))
+
+            # Create a set of unique tokens excluding the flashloan token
+            x_tokens = {token for pair in carbon_pairs for token in pair.split("/") if token != flt}
+
             # Get relevant pairs containing the flashloan token
-            flt_x_pairs = sort_pairs([f"{x}/{flt}" for x in x_tokens])
-            
+            flt_x_pairs = sort_pairs([f"{x_token}/{flt}" for x_token in x_tokens])
+
             # Generate all possible 2-item combinations from the unique tokens that arent the flashloan token
             x_y_pairs = sort_pairs(["{}/{}".format(x, y) for x, y in combinations(x_tokens, 2)])
-            
+
             # Note the relevant pairs
             all_relevant_pairs = flt_x_pairs + x_y_pairs
             self.ConfigObj.logger.debug(f"len(all_relevant_pairs) {len(all_relevant_pairs)}")
@@ -40,21 +40,21 @@ def get_combos(self) -> List[Any]:
 
             # Get pair info for the cohort
             all_relevant_pairs_info = get_all_relevant_pairs_info(self.CCm, all_relevant_pairs, self.ConfigObj.CARBON_V1_FORKS)
-            
+
             # Generate valid triangles for the groups base on arb_mode
             valid_triangles = get_triangle_groups_stats(triangle_groups, all_relevant_pairs_info)
-            
-            # Get [(flt,curves)] analysis set for the flt
+
+            # Get [(flt,curves)] analysis set for the flashloan token
             flt_triangle_analysis_set = get_analysis_set_per_flt(flt, valid_triangles, all_relevant_pairs_info)
-            
-            # The entire analysis set for all flts
+
+            # The entire analysis set for all flashloan tokens
             combos.extend(flt_triangle_analysis_set)
 
         return combos
 
 def sort_pairs(pairs):
     # Clean up the pairs alphabetically
-    return ["/".join(sorted(pair.split('/'))) for pair in pairs]
+    return ["/".join(sorted(pair.split("/"))) for pair in pairs]
 
 def flatten_nested_items_in_list(nested_list):
     # unpack nested items
@@ -73,61 +73,46 @@ def get_triangle_groups(flt, x_y_pairs):
     # Get groups of triangles that conform to (flt/x , x/y, y/flt) where x!=y
     triangle_groups = []
     for pair in x_y_pairs:
-        x,y = pair.split('/')
-        triangle_groups += [("/".join(sorted([flt,x])), pair, "/".join(sorted([flt,y])))]
+        x, y = pair.split("/")
+        triangle_groups += [("/".join(sorted([flt, x])), pair, "/".join(sorted([flt, y])))]
     return triangle_groups
 
 def get_all_relevant_pairs_info(CCm, all_relevant_pairs, carbon_v1_forks):
     # Get pair info for the cohort to allow decision making at the triangle level
     all_relevant_pairs_info = {}
-    for pair in all_relevant_pairs:            
-        all_relevant_pairs_info[pair] = {}
+    for pair in all_relevant_pairs:
         pair_curves = CCm.bypair(pair)
-        carbon_curves = []
-        non_carbon_curves = []
-        base_direction_one = []
-        base_direction_two = []
-        if len(pair_curves) > 0:
-            base_direction_pair = pair_curves[0].pair
-            for x in pair_curves:
-                if x.params.exchange in carbon_v1_forks:
-                    carbon_curves += [x]
-                    if x.pair == base_direction_pair:
-                        base_direction_one += [x]
-                    else:
-                        base_direction_two += [x]
-                else:
-                    non_carbon_curves += [x]
-        all_relevant_pairs_info[pair]['curves'] = non_carbon_curves
-        # for each direction, condense carbon curves into a single list and add to the non-carbon curves
-        if len(base_direction_one) > 0:
-            all_relevant_pairs_info[pair]['curves'].append(base_direction_one)
-        if len(base_direction_two) > 0:
-            all_relevant_pairs_info[pair]['curves'].append(base_direction_two)
-        all_relevant_pairs_info[pair]['all_counts'] = len(pair_curves)
-        all_relevant_pairs_info[pair]['carbon_counts'] = len(carbon_curves)
+        carbon_curves = [curve for curve in pair_curves if curve.params.exchange in carbon_v1_forks]
+        other_curves = [curve for curve in pair_curves if curve.params.exchange not in carbon_v1_forks]
+        all_relevant_pairs_info[pair] = {
+            "has_any": len(pair_curves) > 0,
+            "has_carbon": len(carbon_curves) > 0,
+            "curves": other_curves
+        }
+        if len(carbon_curves) > 0:
+            base_dir_one = [curve for curve in carbon_curves if curve.pair == carbon_curves[0].pair]
+            base_dir_two = [curve for curve in carbon_curves if curve.pair != carbon_curves[0].pair]
+            if len(base_dir_one) > 0:
+                all_relevant_pairs_info[pair]["curves"].append(base_dir_one)
+            if len(base_dir_two) > 0:
+                all_relevant_pairs_info[pair]["curves"].append(base_dir_two)
     return all_relevant_pairs_info
 
 def get_triangle_groups_stats(triangle_groups, all_relevant_pairs_info):
     # Get the stats on the triangle group cohort for decision making
     valid_carbon_triangles = []
     for triangle in triangle_groups:
-        path_len = 0
-        has_carbon = False
-        for pair in triangle:
-            if all_relevant_pairs_info[pair]['all_counts'] > 0:
-                path_len += 1
-            if all_relevant_pairs_info[pair]['carbon_counts'] > 0:
-                has_carbon = True
+        path_len = sum(all_relevant_pairs_info[pair]["has_any"] for pair in triangle)
+        has_carbon = any(all_relevant_pairs_info[pair]["has_carbon"] for pair in triangle)
         if path_len == 3 and has_carbon == True:
             valid_carbon_triangles.append(triangle)
     return valid_carbon_triangles
 
 def get_analysis_set_per_flt(flt, valid_triangles, all_relevant_pairs_info):
     flt_triangle_analysis_set = []
     for triangle in valid_triangles:
-        multiverse = [all_relevant_pairs_info[pair]['curves'] for pair in triangle]
+        multiverse = [all_relevant_pairs_info[pair]["curves"] for pair in triangle]
         product_of_triangle = list(product(multiverse[0], multiverse[1], multiverse[2]))
         triangles_to_run = flatten_nested_items_in_list(product_of_triangle)
-        flt_triangle_analysis_set += list(zip([flt] * len(triangles_to_run), triangles_to_run))        
+        flt_triangle_analysis_set += list(zip([flt] * len(triangles_to_run), triangles_to_run))
     return flt_triangle_analysis_set