# Other library imports from json import loads # Project imports from settings import settings import util class Markets(util.Base): """ " Markets handler for generic market functions. """ def get_all_assets(self, platform): if platform == "agora": assets = loads(settings.Agora.AssetList) elif platform == "lbtc": assets = loads(settings.LocalBitcoins.AssetList) return assets def get_all_providers(self, platform): if platform == "agora": providers = loads(settings.Agora.ProviderList) elif platform == "lbtc": providers = loads(settings.LocalBitcoins.ProviderList) return providers def get_all_currencies(self, platform): if platform == "agora": currencies = list(set([x[0] for x in loads(settings.Agora.DistList)])) elif platform == "lbtc": currencies = list(set([x[0] for x in loads(settings.Agora.DistList)])) return currencies def get_new_ad_equations(self, platform, public_ads, assets=None): """ Update all our prices. :param public_ads: dictionary of public ads keyed by currency :type public_ads: dict :return: list of ads to modify :rtype: list """ if platform == "agora": username = settings.Agora.Username elif platform == "lbtc": username = settings.LocalBitcoins.Username to_update = [] # NOTES: # Get all ads for each currency, with all the payment methods. # Create a function to, in turn, filter these so it contains only one payment method. Run autoprice on this. # Append all results to to_update. Repeat for remaining payment methods, then call slow update. # (asset, currency, provider) if not assets: assets = self.get_all_assets(platform) currencies = self.get_all_currencies(platform) providers = self.get_all_providers(platform) brute = [(asset, currency, provider) for asset in assets for currency in currencies for provider in providers] for asset, currency, provider in brute: # Filter currency try: public_ads_currency = public_ads[currency] except KeyError: # self.log.error("Error getting public ads for currency {currency}", currency=currency) if currency == "GBP": self.log.error("Error getting public ads for currency USD, aborting") break continue # Filter asset public_ads_filtered = [ad for ad in public_ads_currency if ad[4] == asset] # Filter provider public_ads_filtered = [ad for ad in public_ads_filtered if ad[3] == provider] our_ads = [ad for ad in public_ads_filtered if ad[1] == username] if not our_ads: continue new_margin = self.autoprice(public_ads_filtered, currency) # self.log.info("New rate for {currency}: {rate}", currency=currency, rate=new_margin) new_formula = f"coingecko{asset.lower()}usd*usd{currency.lower()}*{new_margin}" for ad in our_ads: ad_id = ad[0] asset = ad[4] our_margin = ad[5] if new_margin != our_margin: to_update.append([ad_id, new_formula, asset, currency, False]) return to_update 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[6]) # 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[6] > 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[6]) # 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[6] # - 0.005 # 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) elif lowball_lowest_not_ours > float(settings.Agora.MaxMargin): # self.log.debug("Lowball lowest not ours more than MaxMargin") return float(settings.Agora.MaxMargin) 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[4]) cheapest_ad_margin = cheapest_ad[6] # - 0.005 if cheapest_ad_margin > float(settings.Agora.MaxMargin): # self.log.debug("Cheapest ad not ours more than MaxMargin") return float(settings.Agora.MaxMargin) # self.log.debug("Cheapest ad above our min that is not us: {x}", x=cheapest_ad) return cheapest_ad_margin def create_distribution_list(self, filter_asset=None): """ Create a list for distribution of ads. :return: generator of asset, countrycode, currency, provider :rtype: generator of tuples """ # Iterate providers like REVOLUT, NATIONAL_BANK for provider in loads(settings.Agora.ProviderList): # Iterate assets like XMR, BTC for asset in loads(settings.Agora.AssetList): # Iterate pairs of currency and country like EUR, GB for currency, countrycode in loads(settings.Agora.DistList): if filter_asset: if asset == filter_asset: yield (asset, countrycode, currency, provider) else: yield (asset, countrycode, currency, provider) def get_valid_account_details(self): currencies = self.sinks.currencies account_info = self.sinks.account_info all_currencies = self.get_all_currencies() supported_currencies = [currency for currency in currencies if currency in all_currencies] currency_account_info_map = {} for currency in supported_currencies: for bank, accounts in account_info.items(): for account in accounts: if account["currency"] == currency: currency_account_info_map[currency] = account["account_number"] return (supported_currencies, currency_account_info_map) def _distribute_account_details(self, platform, currencies=None, account_info=None): """ Distribute account details for ads. We will disable ads we can't support. """ if platform == "agora": caller = self.agora elif platform == "lbtc": caller = self.lbtc if not currencies: currencies = self.sinks.currencies if not account_info: account_info = self.sinks.account_info # First, let's get the ads we can't support all_currencies = self.get_all_currencies() supported_currencies = [currency for currency in currencies if currency in all_currencies] # not_supported = [currency for currency in all_currencies if currency not in supported_currencies] our_ads = caller.enum_ads() supported_ads = [ad for ad in our_ads if ad[3] in supported_currencies] not_supported_ads = [ad for ad in our_ads if ad[3] not in supported_currencies] currency_account_info_map = {} for currency in supported_currencies: for bank, accounts in account_info.items(): for account in accounts: if account["currency"] == currency: currency_account_info_map[currency] = account["account_number"] for ad in supported_ads: asset = ad[0] countrycode = ad[2] currency = ad[3] provider = ad[4] payment_details = currency_account_info_map[currency] ad_id = ad[1] caller.create_ad( asset, countrycode, currency, provider, payment_details, visible=True, edit=True, ad_id=ad_id, ) for ad in not_supported_ads: asset = ad[0] countrycode = ad[2] currency = ad[3] provider = ad[4] ad_id = ad[1] caller.create_ad( asset, countrycode, currency, provider, payment_details=False, visible=False, edit=True, ad_id=ad_id, ) def distribute_account_details(self, currencies=None, account_info=None): """ Helper to distribute the account details for all platforms. """ platforms = ("agora", "lbtc") for platform in platforms: self.distribute_account_details(platform, currencies=currencies, account_info=account_info) def format_ad(self, asset, currency, payment_details_text): """ Format the ad. """ ad = settings.Platform.Ad # Substitute the currency ad = ad.replace("$CURRENCY$", currency) # Substitute the asset ad = ad.replace("$ASSET$", asset) # Substitute the payment details ad = ad.replace("$PAYMENT$", payment_details_text) # Strip extra tabs ad = ad.replace("\\t", "\t") return ad def format_payment_details(self, currency, payment_details): """ Format the payment details. """ payment = settings.Platform.PaymentDetails payment_text = "" for field, value in payment_details.items(): formatted_name = field.replace("_", " ") formatted_name = formatted_name.capitalize() payment_text += f"* {formatted_name}: **{value}**" if field != list(payment_details.keys())[-1]: # No trailing newline payment_text += "\n" payment = payment.replace("$PAYMENT$", payment_text) payment = payment.replace("$CURRENCY$", currency) return payment