Skip to content

Commit

Permalink
Merge pull request #675 from bancorprotocol/arb-mode-code-unification
Browse files Browse the repository at this point in the history
Unify the code across the arb-mode classes
  • Loading branch information
barakman authored May 26, 2024
2 parents 2e64ec1 + f3e45f1 commit 37b804f
Show file tree
Hide file tree
Showing 5 changed files with 97 additions and 115 deletions.
25 changes: 13 additions & 12 deletions fastlane_bot/modes/pairwise_multi_all.py
Original file line number Diff line number Diff line change
Expand Up @@ -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]
21 changes: 10 additions & 11 deletions fastlane_bot/modes/pairwise_multi_pol.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
24 changes: 11 additions & 13 deletions fastlane_bot/modes/triangle_bancor_v3_two_hop.py
Original file line number Diff line number Diff line change
Expand Up @@ -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]

Expand Down
57 changes: 28 additions & 29 deletions fastlane_bot/modes/triangle_multi.py
Original file line number Diff line number Diff line change
Expand Up @@ -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]
85 changes: 35 additions & 50 deletions fastlane_bot/modes/triangle_multi_complete.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)}")
Expand All @@ -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
Expand All @@ -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

0 comments on commit 37b804f

Please sign in to comment.