143 lines
4.4 KiB
Python
143 lines
4.4 KiB
Python
# Other library imports
|
|
from pycoingecko import CoinGeckoAPI
|
|
from forex_python.converter import CurrencyRates
|
|
|
|
# Project imports
|
|
from settings import settings
|
|
import util
|
|
|
|
|
|
class Money(util.Base):
|
|
"""
|
|
Generic class for handling money-related matters that aren't Revolut or Agora.
|
|
"""
|
|
|
|
def __init__(self):
|
|
"""
|
|
Initialise the Money object.
|
|
Set the logger.
|
|
Initialise the CoinGecko API.
|
|
"""
|
|
super().__init__()
|
|
self.cr = CurrencyRates()
|
|
self.cg = CoinGeckoAPI()
|
|
|
|
def lookup_rates(self, ads, rates=None):
|
|
"""
|
|
Lookup the rates for a list of public ads.
|
|
"""
|
|
if not rates:
|
|
rates = self.cg.get_price(
|
|
ids=["monero", "bitcoin"],
|
|
vs_currencies=self.markets.get_all_currencies("agora"),
|
|
)
|
|
# Set the price based on the asset
|
|
for ad in ads:
|
|
if ad[4] == "XMR":
|
|
coin = "monero"
|
|
elif ad[4] == "BTC":
|
|
coin = "bitcoin" # No s here
|
|
currency = ad[5]
|
|
base_currency_price = rates[coin][currency.lower()]
|
|
price = float(ad[2])
|
|
rate = round(price / base_currency_price, 2)
|
|
ad.append(rate)
|
|
# TODO: sort?
|
|
return sorted(ads, key=lambda x: x[2])
|
|
|
|
def get_rates_all(self):
|
|
"""
|
|
Get all rates that pair with USD.
|
|
:return: dictionary of USD/XXX rates
|
|
:rtype: dict
|
|
"""
|
|
rates = self.cr.get_rates("USD")
|
|
return rates
|
|
|
|
def get_acceptable_margins(self, currency, amount):
|
|
"""
|
|
Get the minimum and maximum amounts we would accept a trade for.
|
|
:param currency: currency code
|
|
:param amount: amount
|
|
:return: (min, max)
|
|
:rtype: tuple
|
|
"""
|
|
rates = self.get_rates_all()
|
|
if currency == "USD":
|
|
min_amount = amount - float(settings.Agora.AcceptableUSDMargin)
|
|
max_amount = amount + float(settings.Agora.AcceptableUSDMargin)
|
|
return (min_amount, max_amount)
|
|
amount_usd = amount / rates[currency]
|
|
min_usd = amount_usd - float(settings.Agora.AcceptableUSDMargin)
|
|
max_usd = amount_usd + float(settings.Agora.AcceptableUSDMargin)
|
|
min_local = min_usd * rates[currency]
|
|
max_local = max_usd * rates[currency]
|
|
return (min_local, max_local)
|
|
|
|
def get_minmax(self, asset, currency):
|
|
rates = self.get_rates_all()
|
|
if currency not in rates and not currency == "USD":
|
|
self.log.error(f"Can't create ad without rates: {currency}")
|
|
return
|
|
if asset == "XMR":
|
|
min_usd = float(settings.Agora.MinUSDXMR)
|
|
max_usd = float(settings.Agora.MaxUSDXMR)
|
|
elif asset == "BTC":
|
|
min_usd = float(settings.Agora.MinUSDBTC)
|
|
max_usd = float(settings.Agora.MaxUSDBTC)
|
|
if currency == "USD":
|
|
min_amount = min_usd
|
|
max_amount = max_usd
|
|
else:
|
|
min_amount = rates[currency] * min_usd
|
|
max_amount = rates[currency] * max_usd
|
|
|
|
return (min_amount, max_amount)
|
|
|
|
def to_usd(self, amount, currency):
|
|
if currency == "USD":
|
|
return float(amount)
|
|
else:
|
|
rates = self.get_rates_all()
|
|
return float(amount) / rates[currency]
|
|
|
|
def multiple_to_usd(self, currency_map):
|
|
"""
|
|
Convert multiple curencies to USD while saving API calls.
|
|
"""
|
|
rates = self.get_rates_all()
|
|
cumul = 0
|
|
for currency, amount in currency_map.items():
|
|
if currency == "USD":
|
|
cumul += float(amount)
|
|
else:
|
|
cumul += float(amount) / rates[currency]
|
|
return cumul
|
|
|
|
# TODO: move to money
|
|
def get_profit(self, trades=False):
|
|
"""
|
|
Check how much total profit we have made.
|
|
:return: profit in USD
|
|
:rtype: float
|
|
"""
|
|
total_usd = self.tx.get_total_usd()
|
|
if not total_usd:
|
|
return False
|
|
if trades:
|
|
trades_usd = self.tx.get_open_trades_usd()
|
|
total_usd += trades_usd
|
|
|
|
profit = total_usd - float(settings.Money.BaseUSD)
|
|
if trades:
|
|
cast_es = {
|
|
"profit_trades_usd": profit,
|
|
}
|
|
else:
|
|
cast_es = {
|
|
"profit_usd": profit,
|
|
}
|
|
|
|
self.tx.write_to_es("get_profit", cast_es)
|
|
return profit
|