从两个exchange (Python)获取数据

Question

我正试图从两个交易所得到最新的价格，并根据价格做进一步的行动。

exchanges.py：

from abc import ABC, abstractmethod
from bybit import usdt_perpetual
import requests


class Exchange(ABC):
    @abstractmethod
    def get_latest_price(self):
        pass

class Bybit(Exchange):
    session_unauth = usdt_perpetual.HTTP(endpoint="https://api.bybit.com")

    def __init__(self, symbol: str):
        self.symbol = symbol

    def get_latest_price(self) -> float:
        """
        Get the latest symbol's bid price from Bybit.
        :return: latest bid price
        """
        result = self.session_unauth.orderbook(symbol=self.symbol)["result"]
        return float(result[0]["price"])


class Fin(Exchange):
    url = "fakeurl"

    def __init__(self, symbol):
        self.symbol = symbol[:-1]

    def get_latest_price(self) -> float:
        """
        Get the latest symbol's bid price from the exchange.
        :return: latest bid price
        """
        x = requests.get(self.url)
        return x

app.py：

from exchanges import Fin, Bybit

def main(symbol: str):
    fin = Fin(symbol=symbol)
    bybit = Bybit(symbol=symbol)
    while True:
        fin_price = fin.get_latest_price()
        bybit_price = bybit.get_latest_price()
        # use both price to do futher action...

if __name__ == "__main__":
     main(symbol="BTCUSDT")

有什么方法可以提高代码的可读性和效率？Fin和Bybit似乎非常相似。我能改进这个程序的结构吗？

Answer 1

以下是我的建议。

exchanges.py：

from abc import ABC, abstractmethod
import bybit
import requests


class Exchange(ABC):
    @abstractmethod
    def get_latest_price(self, symbol: str):
        # 1. add the symbol to  params, so you need no separate object
        # instance for every symbol.
        pass


class Bybit(Exchange):
    """Consult price at ByBit exchange."""
    # 2. add class level documentation

    def __init__(self):
        # 3. move session into init, tying it to the object (instance)
        # enabling a refresh/disconnect and easier unit tests.

        # 4. refer explicitly to module, rather than blanket import
        self.session_unauth = bybit.usdt_perpetual.HTTP(
            endpoint="https://api.bybit.com")

    def get_latest_price(self, symbol: str) -> float:
        """
        Get the latest symbol's bid price from Bybit.
        :return: latest bid price
        """
        result = self.session_unauth.orderbook(symbol=symbol)["result"]
        return float(result[0]["price"])


class Fin(Exchange):
    """Consult price at Fin exchange."""
    # see 2.

    FIN_EXCHANGE_URL = "fakeurl"
    # 5. rename constant to uppercase, and descriptive name

    # def __init__(self, symbol):
    #     self.symbol = symbol[:-1]
    # 6. remove unused code.

    def get_latest_price(self, symbol:str) -> float:
        """
        Get the latest symbol's bid price from the exchange.
        :return: latest bid price
        """
        # 7. inline x, removing a non-descriptive variable while reducing LoC
        return requests.get(self.FIN_EXCHANGE_URL)

app.py：

from exchanges import Fin, Bybit

# 8. rename "main" to descriptive method name
def get_symbol_from_multiple_exchanges(symbol: str):
    # 9. since both exchanges are of the same base class, you can
    # refer to them as instances of the base class:
    exchanges = [Fin(), Bybit()]
    while True:
        prices_of_symbol = [exchange.get_latest_price(symbol) for exchange in exchanges]
        # use both price to do futher action...


if __name__ == "__main__":
    # apply 8. here as well
    get_symbol_from_multiple_exchanges(symbol="BTCUSDT")