triangular_trading/Agent.py at master · glitznerf/triangular_trading · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
# A trading agent spotting arbitrage opportunities

from Broker import Broker
import time

class Agent:

    # Initialisation, NOTE: Adjust owned base currency and threshold
    def __init__(self, broker):
        self.broker = broker
        self.base = "USDT"
        self.fees = broker.get_fees()[1]
        self.profit_threshold = 0.0001
        self.file = "opportunities.csv"


    # Print status of the Agent
    def status(self):
        print("Agent Status:")
        print(f"\tBase currency: {self.base}")
        print("\tBroker: OK") if self.broker.ping() else print("\tBroker: Not connected")
        print("\tFees: OK") if len(self.fees) != 0 else print("\tFees: Not set up")
        print(f"\tProfit Threshold: {self.profit_threshold}\n")

    # Calculate the relative outcome of the triangular trade indexed at one
    # Returns float
    def triangular_price(self, direction, order_prices, fees):
        base = 1.0
        for index, dir in enumerate(direction):
            if dir == "buy":
                base = (base * order_prices[index])
            else:
                base = base / order_prices[index]
            base = base - base*fees[index]
        return base

    # Calculate the maximum triangular trade quantities depending on the minimum
    # trade quantity of the three trades available.
    # Returns array of three floats
    def triangular_quantities(self, order_prices, direction, offered_quantities):
        a_1 = offered_quantities[0]*order_prices[1] if direction[1] == "buy" else offered_quantities[0]/order_prices[1]
        b_1 = offered_quantities[1]
        c_1 = offered_quantities[2]/order_prices[2] if direction[2] == "buy" else offered_quantities[2]*order_prices[2]

        if (a_1 < b_1) and (a_1 < c_1):
            a_2 = a_1*order_prices[2] if direction[2] == "buy" else a_1/order_prices[2]
            return [offered_quantities[0], a_1, a_2]
        elif (b_1 < a_1) and (b_1 < c_1):
            b_0 = b_1/order_prices[1] if direction[1] == "buy" else b_1*order_prices[1]
            b_2 = b_1*order_prices[2] if direction[2] == "buy" else b_1/order_prices[2]
            return [b_0, b_1, b_2]
        else:
            c_0 = c_1/order_prices[1] if direction[1] == "buy" else c_1*order_prices[1]
            return [c_0, c_1, offered_quantities[2]]


    # Parse prices from broker for arbitrage opportunities
    # Prints terminal message specifying exact trade opportunities
    def find_arbitrage(self):
        prices = self.broker.get_best_prices()
        logic_start_time = time.time()*1000
        A, B, C = self.base, "", ""
        direction = ["", "", ""]
        pairs = ["", "", ""]
        order_prices = [0, 0, 0]
        fees = [0, 0, 0]
        offered_quantities = [0, 0, 0]
        with open(self.file, "a") as file:
            for i, (B, info) in enumerate(prices[A].items()):
                direction[0] = info[0]
                pairs[0] = info[1]
                order_prices[0] = info[2]
                ix = 0 if direction[0] == "buy" else 1
                fees[0] = self.fees[pairs[0]][ix]
                offered_quantities[0] = info[3]
                try:
                    for i2, (C, info2) in enumerate(prices[B].items()):
                        direction[1] = info2[0]
                        pairs[1] = info2[1]
                        order_prices[1] = info2[2]
                        ix = 0 if direction[1] == "buy" else 1
                        fees[1] = self.fees[pairs[1]][ix]
                        offered_quantities[1] = info[3]

                        CA = prices[C][A]
                        direction[2] = CA[0]
                        pairs[2] = CA[1]
                        order_prices[2] = CA[2]
                        ix = 0 if direction[2] == "buy" else 1
                        fees[2] = self.fees[pairs[2]][ix]
                        offered_quantities[2] = CA[3]

                        effective_price = self.triangular_price(direction, order_prices, fees)
                        if effective_price > 1 + self.profit_threshold:
                            quantities = self.triangular_quantities(order_prices, direction, offered_quantities)
                            logic_end_time = time.time()*1000
                            profit = (effective_price-1.0)*100.0
                            print(f"\tLogic time: {logic_end_time - logic_start_time:.2f} ms")
                            message = ""
                            csv_row = f"{time.time()},{profit},"
                            for i in range(3):
                                message += f"{direction[i]} {quantities[i]} of {pairs[i]} for {order_prices[i]}, "
                                csv_row += f"{pairs[i]},{direction[i]},{quantities[i]},{order_prices[i]},"
                            print(f"For {profit:.4f}% profit, {message}")
                            file.write(csv_row + "\n")

                except KeyError:
                    pass

    # Get dictionary of ticker prices by symbol
    # Returns: either (True, prices (dict)) or (False, False)
    def clean_prices(self):
        raw = self.broker.get_price()
        if raw[0]:
            prices = {}
            for entry in raw[1]:
                prices[entry["symbol"]] = float(entry["price"])
            return (True, prices)
        else:
            return (False, False)