pluto/handler/agora.py

# Twisted/Klein imports
from twisted.logger import Logger
from twisted.internet.task import LoopingCall
from twisted.internet.threads import deferToThread

# Other library imports
from json import loads
from forex_python.converter import CurrencyRates
from agoradesk_py import AgoraDesk
from httpx import ReadTimeout, ReadError
from pycoingecko import CoinGeckoAPI
from datetime import datetime

# Project imports
from settings import settings


class Agora(object):
    """
    AgoraDesk API handler.
    """

    def __init__(self):
        """
        Initialise the AgoraDesk and CurrencyRates APIs.
        Initialise the last_dash storage for detecting new trades.
        """
        self.log = Logger("agora")
        self.agora = AgoraDesk(settings.Agora.Token)
        self.cr = CurrencyRates()
        self.cg = CoinGeckoAPI()

        # Cache for detecting new trades
        self.last_dash = set()

        # Cache for detecting new messages
        self.last_messages = {}

    def set_irc(self, irc):
        self.irc = irc

    def set_tx(self, tx):
        self.tx = tx

    def setup_loop(self):
        """
        Set up the LoopingCall to get all active trades and messages.
        """
        self.lc_dash = LoopingCall(self.loop_check)
        self.lc_dash.start(int(settings.Agora.RefreshSec))
        if settings.Agora.Cheat == "1":
            self.lc_cheat = LoopingCall(self._update_prices)
            self.lc_cheat.start(int(settings.Agora.CheatSec))

    def wrap_dashboard(self):
        try:
            dash = self.agora.dashboard_seller()
        except ReadTimeout:
            return False
        if not dash:
            return False
        dash_tmp = {}
        if "data" not in dash["response"].keys():
            self.log.error("Data not in dashboard response: {content}", content=dash)
            return dash_tmp
        if dash["response"]["data"]["contact_count"] > 0:
            for contact in dash["response"]["data"]["contact_list"]:
                contact_id = contact["data"]["contact_id"]
                dash_tmp[contact_id] = contact
        return dash_tmp

    def loop_check(self):
        """
        Calls hooks to parse dashboard info and get all contact messages.
        """
        dash_tmp = self.wrap_dashboard()

        # Call dashboard hooks
        self.dashboard_hook(dash_tmp)

        # Get recent messages
        try:
            self.get_recent_messages()
        except ReadTimeout:
            pass
        return dash_tmp

    def get_dashboard(self):
        """
        Get dashboard helper for IRC only.
        """
        dash = self.wrap_dashboard()
        rtrn = []
        if not dash.items():
            return
        for contact_id, contact in dash.items():
            reference = self.tx.tx_to_ref(contact_id)
            buyer = contact["data"]["buyer"]["username"]
            amount = contact["data"]["amount"]
            asset = contact["data"]["advertisement"]["asset"]
            if asset == "XMR":
                amount_crypto = contact["data"]["amount_xmr"]
            elif asset == "BTC":
                amount_crypto = contact["data"]["amount_btc"]
            currency = contact["data"]["currency"]
            if not contact["data"]["is_selling"]:
                continue
            rtrn.append(f"{reference}: {buyer} {amount}{currency} {amount_crypto}{asset}")
        return rtrn

    def dashboard_hook(self, dash):
        """
        Get information about our open trades.
        Post new trades to IRC and cache trades for the future.
        """
        current_trades = []
        if not dash:
            return
        if not dash.items():
            return
        for contact_id, contact in dash.items():
            reference = self.tx.tx_to_ref(contact_id)
            if reference:
                current_trades.append(reference)
            buyer = contact["data"]["buyer"]["username"]
            amount = contact["data"]["amount"]
            asset = contact["data"]["advertisement"]["asset"]
            if asset == "XMR":
                amount_crypto = contact["data"]["amount_xmr"]
            elif asset == "BTC":
                amount_crypto = contact["data"]["amount_btc"]
            currency = contact["data"]["currency"]
            if not contact["data"]["is_selling"]:
                continue
            if reference not in self.last_dash:
                reference = self.tx.new_trade(asset, contact_id, buyer, currency, amount, amount_crypto)
                if reference:
                    if reference not in current_trades:
                        current_trades.append(reference)
                # Let us know there is a new trade
                self.irc.sendmsg(f"AUTO {reference}: {buyer} {amount}{currency} {amount_crypto}{asset}")
            # Note that we have seen this reference
            self.last_dash.add(reference)

        # Purge old trades from cache
        for ref in list(self.last_dash):  # We're removing from the list on the fly
            if ref not in current_trades:
                self.last_dash.remove(ref)
        if reference and reference not in current_trades:
            current_trades.append(reference)
        self.tx.cleanup(current_trades)

    def dashboard_release_urls(self):
        """
        Get information about our open trades.
        Post new trades to IRC and cache trades for the future.
        :return: human readable list of strings about our trades or False
        :rtype: list or bool
        """
        dash = self.agora.dashboard_seller()
        dash_tmp = []
        if "data" not in dash["response"]:
            self.log.error("Data not in dashboard response: {content}", content=dash)
            return False
        if dash["response"]["data"]["contact_count"] > 0:
            for contact in dash["response"]["data"]["contact_list"]:
                contact_id = contact["data"]["contact_id"]
                buyer = contact["data"]["buyer"]["username"]
                amount = contact["data"]["amount"]
                asset = contact["data"]["advertisement"]["asset"]
                if asset == "XMR":
                    amount_crypto = contact["data"]["amount_xmr"]
                elif asset == "BTC":
                    amount_crypto = contact["data"]["amount_btc"]
                currency = contact["data"]["currency"]
                release_url = contact["actions"]["release_url"]
                if not contact["data"]["is_selling"]:
                    continue
                reference = self.tx.tx_to_ref(contact_id)
                if not reference:
                    reference = "not_set"
                dash_tmp.append(f"{reference}: {buyer} {amount}{currency} {amount_crypto}{asset} {release_url}")

        return dash_tmp

    def get_recent_messages(self, send_irc=True):
        """
        Get recent messages.
        """
        messages_tmp = {}
        try:
            messages = self.agora.recent_messages()
        except ReadError:
            return False
        if not messages["success"]:
            return False
        if "data" not in messages["response"]:
            self.log.error("Data not in messages response: {content}", content=messages["response"])
            return False
        open_tx = self.tx.get_ref_map().keys()
        for message in messages["response"]["data"]["message_list"]:
            contact_id = message["contact_id"]
            username = message["sender"]["username"]
            msg = message["msg"]
            if contact_id not in open_tx:
                continue
            reference = self.tx.tx_to_ref(contact_id)
            if reference in messages_tmp:
                messages_tmp[reference].append([username, msg])
            else:
                messages_tmp[reference] = [[username, msg]]

            # Send new messages on IRC
            if send_irc:
                for user, message in messages_tmp[reference]:
                    if reference in self.last_messages:
                        if not [user, message] in self.last_messages[reference]:
                            self.irc.sendmsg(f"AUTO {reference}: ({user}) {message}")
                            # Append sent messages to last_messages so we don't send them again
                            self.last_messages[reference].append([user, message])
                    else:
                        self.last_messages[reference] = [[user, message]]
                        for x in messages_tmp[reference]:
                            self.irc.sendmsg(f"NEW {reference}: ({user}) {message}")

        # Purge old trades from cache
        for ref in list(self.last_messages):  # We're removing from the list on the fly
            if ref not in messages_tmp:
                del self.last_messages[ref]

        return messages_tmp

    def enum_ad_ids(self, page=0):
        ads = self.agora._api_call(api_method="ads", query_values={"page": page})
        ads_total = []
        if not ads["success"]:
            return False
        for ad in ads["response"]["data"]["ad_list"]:
            ads_total.append(ad["data"]["ad_id"])
        if "pagination" in ads["response"]:
            if "next" in ads["response"]["pagination"]:
                page += 1
                for ad in self.enum_ad_ids(page):
                    ads_total.append(ad)
        return ads_total

    def enum_ads(self, page=0):
        ads = self.agora._api_call(api_method="ads", query_values={"page": page})
        ads_total = []
        if not ads["success"]:
            return False
        for ad in ads["response"]["data"]["ad_list"]:
            ads_total.append([ad["data"]["asset"], ad["data"]["ad_id"], ad["data"]["countrycode"], ad["data"]["currency"]])
        if "pagination" in ads["response"]:
            if "next" in ads["response"]["pagination"]:
                page += 1
                for ad in self.enum_ads(page):
                    ads_total.append([ad[0], ad[1], ad[2], ad[3]])
        return ads_total

    def last_online_recent(self, date):
        """
        Check if the last online date was recent.
        :param date: date last online
        :type date: string
        :return: bool indicating whether the date was recent enough
        :rtype: bool
        """
        date_parsed = datetime.strptime(date, "%Y-%m-%dT%H:%M:%S.%fZ")
        now = datetime.now()
        sec_ago_date = (now - date_parsed).total_seconds()
        self.log.debug("Seconds ago date for {date} ^ {now}: {x}", date=date, now=str(now), x=sec_ago_date)
        return sec_ago_date < 172800

    def enum_public_ads(self, coin, currency, page=0):
        # buy-monero-online, buy-bitcoin-online
        # Work around Agora weirdness calling it bitcoins
        if coin == "bitcoin":
            coin = "bitcoins"
        ads = self.agora._api_call(api_method=f"buy-{coin}-online/{currency}/REVOLUT", query_values={"page": page})
        if not ads["success"]:
            return False
        for ad in ads["response"]["data"]["ad_list"]:
            if ad["data"]["online_provider"] == "REVOLUT":
                date_last_seen = ad["data"]["profile"]["last_online"]
                # Check if this person was seen recently
                if not self.last_online_recent(date_last_seen):
                    continue
                yield [ad["data"]["ad_id"], ad["data"]["profile"]["username"], ad["data"]["temp_price"]]
        if "pagination" in ads["response"]:
            if "next" in ads["response"]["pagination"]:
                page += 1
                for ad in self.enum_public_ads(coin, currency, page):
                    yield [ad[0], ad[1], ad[2]]

    def wrap_public_ads(self, asset, currency, rates=None):
        """
        Wrapper to sort public ads.
        """
        if asset == "XMR":
            coin = "monero"
        elif asset == "BTC":
            coin = "bitcoin"
        ads = list(self.enum_public_ads(coin, currency.upper()))
        if not rates:
            # Set the price based on the asset
            base_currency_price = self.cg.get_price(ids=coin, vs_currencies=currency)[coin][currency.lower()]
        else:
            base_currency_price = rates
        for ad in ads:
            price = float(ad[2])
            rate = round(price / base_currency_price, 2)
            ad.append(rate)
        return ads

    def _update_prices(self, xmr=True, btc=True):
        """
        Update prices in another thread.
        """
        deferToThread(self.update_prices, xmr, btc)

    def update_prices(self, xmr=True, btc=True):
        our_ads = self.enum_ads()
        currencies = [x[3].lower() for x in our_ads]
        public_ad_dict_xmr = {}
        public_ad_dict_btc = {}
        rates_crypto = self.cg.get_price(ids=["monero", "bitcoin"], vs_currencies=currencies)
        for currency in currencies:
            try:
                rates_xmr = rates_crypto["monero"][currency]
                rates_btc = rates_crypto["bitcoin"][currency]
                if xmr:
                    public_ad_dict_xmr[currency] = self.wrap_public_ads("XMR", currency, rates=rates_xmr)
                if btc:
                    public_ad_dict_btc[currency] = self.wrap_public_ads("BTC", currency, rates=rates_btc)
            except KeyError:
                self.log.error("Error getting public ads for currency {currency}", currency=currency)
                continue
        for asset, ad_id, country, currency in our_ads:
            # Get the ads for this country/currency pair
            try:
                if asset == "XMR":
                    public_ads = public_ad_dict_xmr[currency.lower()]
                elif asset == "BTC":
                    public_ads = public_ad_dict_btc[currency.lower()]
            except KeyError:
                continue
            if not public_ads:
                continue
            if xmr:
                if asset == "XMR":
                    new_margin = self.autoprice(public_ads, currency)
                    new_formula = f"coingeckoxmrusd*usd{currency.lower()}*{new_margin}"
            if btc:
                if asset == "BTC":
                    new_margin = self.autoprice(public_ads, currency)
                    new_formula = f"coingeckobtcusd*usd{currency.lower()}*{new_margin}"
            rtrn = self.agora.ad_equation(ad_id, new_formula)
            if not rtrn["success"]:
                self.log.error("Error updating ad {ad_id}: {response}", ad_id=ad_id, response=rtrn["response"])
            self.log.info("Rate for {currency}: {margin}", currency=currency, margin=new_margin)

    def autoprice(self, ads, currency):
        """
        Helper function to automatically adjust the price up/down in certain markets
        in order to gain the most profits and sales.
        :param ads: list of ads
        :type ads: list of lists
        :param currency: currency of the ads
        :type currency: string
        :return: the rate we should use for this currency
        :rtype: float
        """
        self.log.debug("Autoprice starting for {x}", x=currency)
        # Find cheapest ad
        # Filter by 3rd index on each ad list to find the cheapest
        min_margin_ad = min(ads, key=lambda x: x[3])
        self.log.debug("Minimum margin ad: {x}", x=min_margin_ad)

        # Find second cheapest that is not us
        # Remove results from ads that are us
        ads_without_us = [ad for ad in ads if not ad[1] == settings.Agora.Username]
        self.log.debug("Ads without us: {x}", x=ads_without_us)
        # Find ads above our min that are not us
        ads_above_our_min_not_us = [ad for ad in ads_without_us if ad[3] > float(settings.Agora.MinMargin)]
        self.log.debug("Ads above our min not us: {x}", x=ads_above_our_min_not_us)
        # Check that this list without us is not empty
        if ads_without_us:
            # Find the cheapest from these
            min_margin_ad_not_us = min(ads_without_us, key=lambda x: x[3])
            self.log.debug("Min margin ad not us: {x}", x=min_margin_ad_not_us)
            # Lowball the lowest ad that is not ours
            lowball_lowest_not_ours = min_margin_ad_not_us[3] - 0.001
            self.log.debug("Lowball lowest not ours: {x}", x=lowball_lowest_not_ours)

        # Check if the username field of the cheapest ad matches ours
        if min_margin_ad[1] == settings.Agora.Username:
            self.log.debug("We are the cheapest for: {x}", x=currency)
            # We are the cheapest!
            # Are all of the ads ours?
            all_ads_ours = all([ad[1] == settings.Agora.Username for ad in ads])
            if all_ads_ours:
                self.log.debug("All ads are ours for: {x}", x=currency)
                # Now we know it's safe to return the maximum value
                return float(settings.Agora.MaxMargin)
            else:
                self.log.debug("All ads are NOT ours for: {x}", x=currency)
                # All the ads are not ours, but we are first...
                # Check if the lowballed, lowest (that is not ours) ad's margin
                # is less than our minimum
                if lowball_lowest_not_ours < float(settings.Agora.MinMargin):
                    self.log.debug("Lowball lowest not ours less than MinMargin")
                    return float(settings.Agora.MinMargin)
                else:
                    self.log.debug("Returning lowballed figure: {x}", x=lowball_lowest_not_ours)
                    return lowball_lowest_not_ours
        else:
            self.log.debug("We are NOT the cheapest for: {x}", x=currency)
            # We are not the cheapest :(
            # Check if this list is empty
            if not ads_above_our_min_not_us:
                # Return the maximum margin?
                return float(settings.Agora.MaxMargin)
            # Find cheapest ad above our min that is not us
            cheapest_ad = min(ads_above_our_min_not_us, key=lambda x: x[3])
            self.log.debug("Cheapest ad above our min that is not us: {x}", x=cheapest_ad)
            return cheapest_ad[3] - 0.001

    def nuke_ads(self):
        """
        Delete all of our adverts.
        :return: True or False
        :rtype: bool
        """
        ads = self.enum_ad_ids()
        return_ids = []
        if not ads:
            return False
        for ad_id in ads:
            rtrn = self.agora.ad_delete(ad_id)
            return_ids.append(rtrn["success"])
        return all(return_ids)

    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 create_ad(self, asset, countrycode, currency):
        """
        Post an ad with the given asset in a country with a given currency.
        Convert the min and max amounts from settings to the given currency with CurrencyRates.
        :param asset: the crypto asset to list (XMR or BTC)
        :type asset: string
        :param countrycode: country code
        :param currency: currency code
        :type countrycode: string
        :type currency: string
        :return: data about created object or error
        :rtype: dict
        """
        ad = settings.Agora.Ad
        paymentdetails = settings.Agora.PaymentDetails

        # Substitute the currency
        ad = ad.replace("$CURRENCY$", currency)
        if countrycode == "GB" and currency == "GBP":
            ad = ad.replace("$PAYMENT$", settings.Agora.GBPDetailsAd)
            paymentdetailstext = paymentdetails.replace("$PAYMENT$", settings.Agora.GBPDetailsPayment)
        else:
            ad = ad.replace("$PAYMENT$", settings.Agora.DefaultDetailsAd)
            paymentdetailstext = paymentdetails.replace("$PAYMENT$", settings.Agora.DefaultDetailsPayment)

        # Substitute the asset
        ad = ad.replace("$ASSET$", asset)

        rates = self.get_rates_all()
        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
        price_formula = f"coingecko{asset.lower()}usd*usd{currency.lower()}*{settings.Agora.Margin}"
        # price_formula = f"coingeckoxmrusd*{settings.Agora.Margin}"

        # Remove extra tabs
        ad = ad.replace("\\t", "\t")

        form = {
            "country_code": countrycode,
            "currency": currency,
            "trade_type": "ONLINE_SELL",
            "asset": asset,
            "price_equation": price_formula,
            "track_max_amount": False,
            "require_trusted_by_advertiser": False,
            "online_provider": "REVOLUT",
            "msg": ad,
            "min_amount": min_amount,
            "max_amount": max_amount,
            "payment_method_details": settings.Agora.PaymentMethodDetails,
            "account_info": paymentdetailstext,
        }

        # Dirty hack to test
        # if asset == "BTC":
        #    del form["min_amount"]
        ad = self.agora.ad_create(**form)
        return ad

    def dist_countries(self):
        """
        Distribute our advert into all countries listed in the config.
        Exits on errors.
        :return: False or dict with response
        :rtype: bool or dict
        """
        for asset in loads(settings.Agora.AssetList):
            for currency, countrycode in loads(settings.Agora.DistList):
                rtrn = self.create_ad(asset, countrycode, currency)
                if not rtrn:
                    return False
                yield rtrn

    #   def get_combinations(self):
    #       """
    #       Get all combinations of currencies and countries from the configuration.
    #       :return: list of [country, currency]
    #       :rtype: list
    #       """
    #       currencies = loads(settings.Agora.BruteCurrencies)
    #       countries = loads(settings.Agora.BruteCountries)
    #       combinations = [[country, currency] for country in countries for currency in currencies]
    #       return combinations

    #    def dist_bruteforce(self):
    #        """
    #        Bruteforce all possible ads from the currencies and countries in the config.
    #        Does not exit on errors.
    #        :return: False or dict with response
    #        :rtype: bool or dict
    #        """
    #        combinations = self.get_combinations()
    #        for country, currency in combinations:
    #            rtrn = self.create_ad(country, currency)
    #            if not rtrn:
    #                yield False
    #            yield rtrn
    #
    #    def bruteforce_fill_blanks(self):
    #        """
    #        Get the ads that we want to configure but have not, and fill in the blanks.
    #        :return: False or dict with response
    #        :rtype: bool or dict
    #        """
    #        existing_ads = self.enum_ads()
    #        combinations = self.get_combinations()
    #        for country, currency in combinations:
    #            if not [country, currency] in existing_ads:
    #                rtrn = self.create_ad(country, currency)
    #                if not rtrn:
    #                    yield False
    #                yield rtrn

    def strip_duplicate_ads(self):
        """
        Remove duplicate ads.
        :return: list of duplicate ads
        :rtype: list
        """
        existing_ads = self.enum_ads()
        _size = len(existing_ads)
        repeated = []
        for i in range(_size):
            k = i + 1
            for j in range(k, _size):
                if existing_ads[i] == existing_ads[j] and existing_ads[i] not in repeated:
                    repeated.append(existing_ads[i])

        actioned = []
        for ad_id, country, currency in repeated:
            rtrn = self.agora.ad_delete(ad_id)
            actioned.append(rtrn["success"])
        return all(actioned)

    def release_funds(self, contact_id):
        """
        Release funds for a contact_id.
        :param contact_id: trade/contact ID
        :type contact_id: string
        :return: response dict
        :rtype: dict
        """
        payload = {"tradeId": contact_id, "password": settings.Agora.Pass}
        rtrn = self.agora._api_call(api_method=f"contact_release/{contact_id}", http_method="POST", query_values=payload)

        # Check if we can withdraw funds
        self.withdraw_funds()

        return rtrn

    def withdraw_funds(self):
        """
        Withdraw excess funds to our XMR/BTC wallets.
        """
        totals_all = self.tx.get_total()

        wallet_xmr, _ = totals_all[2]

        # Get the wallet balances in USD
        total_usd = totals_all[0][1]

        profit_usd = total_usd - float(settings.Money.BaseUSD)

        # Get the XMR -> USD exchange rate
        xmr_usd = self.cg.get_price(ids="monero", vs_currencies=["USD"])

        # Convert the USD total to XMR
        profit_usd_in_xmr = float(profit_usd) / xmr_usd["monero"]["usd"]

        # Check profit is above zero
        if not profit_usd >= 0:
            return

        if not float(wallet_xmr) > profit_usd_in_xmr:
            # Not enough funds to withdraw
            self.log.error(
                "Not enough funds to withdraw {profit}, as wallet only contains {wallet}", profit=profit_usd_in_xmr, wallet=wallet_xmr
            )
            self.irc.sendmsg(f"Not enough funds to withdraw {profit_usd_in_xmr}, as wallet only contains {wallet_xmr}")
            return

        if not profit_usd >= float(settings.Money.WithdrawLimit):
            # Not enough profit to withdraw
            return

        half = profit_usd_in_xmr / 2

        half_rounded = round(half, 8)

        # Set up the format for calling wallet_send_xmr
        send_cast = {
            "address": None,
            "amount": half_rounded,
            "password": settings.Agora.Pass,
        }

        send_cast["address"] = settings.XMR.Wallet1
        rtrn1 = self.agora.wallet_send_xmr(**send_cast)

        send_cast["address"] = settings.XMR.Wallet2
        rtrn2 = self.agora.wallet_send_xmr(**send_cast)

        self.irc.sendmsg(f"Withdrawal: {rtrn1['success']} | {rtrn2['success']}")