# Other library imports from json import loads # Project imports from settings import settings import util import db class Markets(util.Base): """ " Markets handler for generic market functions. """ def find_trades_by_uid(self, uid): """ Find a list of trade IDs and references by a customer UID. :return: tuple of (platform, trade_id, reference, currency) """ platform, username = self.ux.verify.get_uid(uid) refs = db.get_refs() matching_trades = [] for reference in refs: ref_data = db.get_ref(reference) tx_platform = ref_data["subclass"] tx_username = ref_data["buyer"] trade_id = ref_data["id"] currency = ref_data["currency"] if tx_platform == platform and tx_username == username: to_append = (platform, trade_id, reference, currency) matching_trades.append(to_append) return matching_trades def get_send_settings(self, platform): if platform == "agora": send_setting = settings.Agora.Send post_message = self.agora.api.contact_message_post elif platform == "lbtc": send_setting = settings.LocalBitcoins.Send post_message = self.lbtc.api.contact_message_post return (send_setting, post_message) def send_reference(self, platform, trade_id, reference): """ Send the reference to a customer. """ send_setting, post_message = self.get_send_settings(platform) if send_setting == "1": post_message( trade_id, f"When sending the payment please use reference code: {reference}", ) def send_bank_details(self, platform, currency, trade_id): """ Send the bank details to a trade. """ send_setting, post_message = self.get_send_settings(platform) self.log.info(f"Sending bank details/reference for {platform}/{trade_id}") if send_setting == "1": account_info = self.get_matching_account_details(platform, currency) formatted_account_info = self.format_payment_details(currency, account_info, real=True) post_message( trade_id, f"Payment details: \n{formatted_account_info}", ) def get_all_assets(self, platform): sets = util.get_settings(platform) assets = loads(sets.AssetList) return assets def get_all_providers(self, platform): sets = util.get_settings(platform) providers = loads(sets.ProviderList) return providers def get_all_currencies(self, platform): sets = util.get_settings(platform) currencies = list(set([x[0] for x in loads(sets.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 """ sets = util.get_settings(platform) username = sets.Username min_margin = sets.MinMargin max_margin = sets.MaxMargin 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) if platform == "lbtc": providers = [self.sources.lbtc.map_provider(x, reverse=True) for x in providers] sinks_currencies = self.sinks.currencies supported_currencies = [currency for currency in currencies if currency in sinks_currencies] currencies = supported_currencies 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: self.log.warning(f"No ads found in {platform} public listing for {asset} {currency} {provider}") continue new_margin = self.autoprice(username, min_margin, max_margin, public_ads_filtered, currency) # self.log.info("New rate for {currency}: {rate}", currency=currency, rate=new_margin) if platform == "agora": new_formula = f"coingecko{asset.lower()}usd*usd{currency.lower()}*{new_margin}" elif platform == "lbtc": new_formula = f"btc_in_usd*{new_margin}*USD_in_{currency}" 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, username, min_margin, max_margin, 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] == 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(min_margin)] # 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] == 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] == 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(max_margin) 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(min_margin): # self.log.debug("Lowball lowest not ours less than MinMargin") return float(min_margin) elif lowball_lowest_not_ours > float(max_margin): # self.log.debug("Lowball lowest not ours more than MaxMargin") return float(max_margin) 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(max_margin) # 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(max_margin): # self.log.debug("Cheapest ad not ours more than MaxMargin") return float(max_margin) # 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, platform, filter_asset=None): """ Create a list for distribution of ads. :return: generator of asset, countrycode, currency, provider :rtype: generator of tuples """ sets = util.get_settings(platform) # Iterate providers like REVOLUT, NATIONAL_BANK for provider in loads(sets.ProviderList): # Iterate assets like XMR, BTC for asset in loads(sets.AssetList): # Iterate pairs of currency and country like EUR, GB for currency, countrycode in loads(sets.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, platform): currencies = self.sinks.currencies account_info = self.sinks.account_info all_currencies = self.get_all_currencies(platform) 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"] currency_account_info_map[currency]["bank"] = bank.split("_")[0] currency_account_info_map[currency]["recipient"] = account["recipient"] return (supported_currencies, currency_account_info_map) def get_matching_account_details(self, platform, currency): supported_currencies, currency_account_info_map = self.get_valid_account_details(platform) if currency not in supported_currencies: return False return currency_account_info_map[currency] 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 supported_currencies, currency_account_info_map = self.get_valid_account_details(platform) # 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] 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, real=False): """ Format the payment details. """ if not payment_details: return False if real: payment = settings.Platform.PaymentDetailsReal else: 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