From 0f3dba2b06f702a7ae98638a18f1893fb84439ed Mon Sep 17 00:00:00 2001 From: Mark Veidemanis Date: Wed, 12 Jan 2022 19:17:53 +0000 Subject: [PATCH] Reimplement autoprice function better --- handler/agora.py | 138 +++++++++++++++++++++++++++-------------------- 1 file changed, 80 insertions(+), 58 deletions(-) diff --git a/handler/agora.py b/handler/agora.py index ace4001..f72c990 100644 --- a/handler/agora.py +++ b/handler/agora.py @@ -8,6 +8,7 @@ from forex_python.converter import CurrencyRates from agoradesk_py import AgoraDesk from httpx import ReadTimeout from pycoingecko import CoinGeckoAPI +from datetime import datetime # Project imports from settings import settings @@ -173,7 +174,7 @@ class Agora(object): messages = self.agora.recent_messages() if not messages["success"]: return False - if not "data" in messages["response"]: + 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() @@ -237,12 +238,30 @@ class Agora(object): ads_total.append([ad[0], ad[1], ad[2]]) 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, currency, page=0): ads = self.agora._api_call(api_method=f"buy-monero-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"]: @@ -263,20 +282,8 @@ class Agora(object): price = float(ad[2]) rate = round(price / base_monero_price, 2) ad.append(rate) - ads.sort(key=lambda x: float(x[2])) return ads - def sort_ads_annotate_place(self, ads): - ads.sort(key=lambda x: float(x[2])) - place = None - for index, ad in enumerate(ads): - if ad[1] == settings.Agora.Username: - place = index - break - if place is None: - print("Place is none for ads:", ads) - return (ads, place) - def update_prices(self): our_ads = self.enum_ads() currencies = [x[2].lower() for x in our_ads] @@ -306,52 +313,67 @@ class Agora(object): """ Helper function to automatically adjust the price up/down in certain markets in order to gain the most profits and sales. - """ - ads, place = self.sort_ads_annotate_place(ads) - all_results_us = all([ad[1] == settings.Agora.Username for ad in ads]) - if place == 0: - if len(ads) > 1 and not all_results_us: - competitor_index = None - for index, ad in enumerate(ads): - if ad[1] != settings.Agora.Username: - competitor_index = index - print("Found a competitor at index", competitor_index, ad) - break - else: - print("Skipping ad", ad) - if not competitor_index: - print("Couldn't find a competitor from ads:", ads) - return float(ads[0][3]) - - next_max = float(ads[competitor_index][3]) - our_max_adjusted = next_max - 0.01 - if our_max_adjusted < float(settings.Agora.MinMargin): - our_max_adjusted = float(settings.Agora.MinMargin) - if our_max_adjusted > float(settings.Agora.MaxMargin): - our_max_adjusted = float(settings.Agora.MaxMargin) - # Lowball next competitor - self.log.info( - "Lowballing next competitor for {currency}: {margin}", - currency=currency, - margin=our_max_adjusted, - ) - return our_max_adjusted + :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: - our_max_adjusted = float(settings.Agora.MaxMargin) - # Set the max rates as people have no choice - return our_max_adjusted - iter_count = 0 - while place > 1 and not iter_count > 100: - iter_count += 1 - recommended_margin = float(ads[place][3]) - 0.01 - if recommended_margin <= float(settings.Agora.MinMargin): - break - ads[place][3] = recommended_margin - print("Set recommended margin for", currency, recommended_margin) - - ads, place = self.sort_ads_annotate_place(ads) - # Return our adjusted (or left) margin - return float(ads[place][3]) + 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): """