Finish implementation and tests for the cheat system #3
134
handler/agora.py
134
handler/agora.py
|
@ -8,6 +8,7 @@ from forex_python.converter import CurrencyRates
|
||||||
from agoradesk_py import AgoraDesk
|
from agoradesk_py import AgoraDesk
|
||||||
from httpx import ReadTimeout
|
from httpx import ReadTimeout
|
||||||
from pycoingecko import CoinGeckoAPI
|
from pycoingecko import CoinGeckoAPI
|
||||||
|
from datetime import datetime
|
||||||
|
|
||||||
# Project imports
|
# Project imports
|
||||||
from settings import settings
|
from settings import settings
|
||||||
|
@ -173,7 +174,7 @@ class Agora(object):
|
||||||
messages = self.agora.recent_messages()
|
messages = self.agora.recent_messages()
|
||||||
if not messages["success"]:
|
if not messages["success"]:
|
||||||
return False
|
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"])
|
self.log.error("Data not in messages response: {content}", content=messages["response"])
|
||||||
return False
|
return False
|
||||||
open_tx = self.tx.get_ref_map().keys()
|
open_tx = self.tx.get_ref_map().keys()
|
||||||
|
@ -237,12 +238,30 @@ class Agora(object):
|
||||||
ads_total.append([ad[0], ad[1], ad[2]])
|
ads_total.append([ad[0], ad[1], ad[2]])
|
||||||
return ads_total
|
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):
|
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})
|
ads = self.agora._api_call(api_method=f"buy-monero-online/{currency}/REVOLUT", query_values={"page": page})
|
||||||
if not ads["success"]:
|
if not ads["success"]:
|
||||||
return False
|
return False
|
||||||
for ad in ads["response"]["data"]["ad_list"]:
|
for ad in ads["response"]["data"]["ad_list"]:
|
||||||
if ad["data"]["online_provider"] == "REVOLUT":
|
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"]]
|
yield [ad["data"]["ad_id"], ad["data"]["profile"]["username"], ad["data"]["temp_price"]]
|
||||||
if "pagination" in ads["response"]:
|
if "pagination" in ads["response"]:
|
||||||
if "next" in ads["response"]["pagination"]:
|
if "next" in ads["response"]["pagination"]:
|
||||||
|
@ -263,20 +282,8 @@ class Agora(object):
|
||||||
price = float(ad[2])
|
price = float(ad[2])
|
||||||
rate = round(price / base_monero_price, 2)
|
rate = round(price / base_monero_price, 2)
|
||||||
ad.append(rate)
|
ad.append(rate)
|
||||||
ads.sort(key=lambda x: float(x[2]))
|
|
||||||
return ads
|
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):
|
def update_prices(self):
|
||||||
our_ads = self.enum_ads()
|
our_ads = self.enum_ads()
|
||||||
currencies = [x[2].lower() for x in our_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
|
Helper function to automatically adjust the price up/down in certain markets
|
||||||
in order to gain the most profits and sales.
|
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
|
||||||
"""
|
"""
|
||||||
ads, place = self.sort_ads_annotate_place(ads)
|
self.log.debug("Autoprice starting for {x}", x=currency)
|
||||||
all_results_us = all([ad[1] == settings.Agora.Username for ad in ads])
|
# Find cheapest ad
|
||||||
if place == 0:
|
# Filter by 3rd index on each ad list to find the cheapest
|
||||||
if len(ads) > 1 and not all_results_us:
|
min_margin_ad = min(ads, key=lambda x: x[3])
|
||||||
competitor_index = None
|
self.log.debug("Minimum margin ad: {x}", x=min_margin_ad)
|
||||||
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])
|
# Find second cheapest that is not us
|
||||||
our_max_adjusted = next_max - 0.01
|
# Remove results from ads that are us
|
||||||
if our_max_adjusted < float(settings.Agora.MinMargin):
|
ads_without_us = [ad for ad in ads if not ad[1] == settings.Agora.Username]
|
||||||
our_max_adjusted = float(settings.Agora.MinMargin)
|
self.log.debug("Ads without us: {x}", x=ads_without_us)
|
||||||
if our_max_adjusted > float(settings.Agora.MaxMargin):
|
# Find ads above our min that are not us
|
||||||
our_max_adjusted = float(settings.Agora.MaxMargin)
|
ads_above_our_min_not_us = [ad for ad in ads_without_us if ad[3] > float(settings.Agora.MinMargin)]
|
||||||
# Lowball next competitor
|
self.log.debug("Ads above our min not us: {x}", x=ads_above_our_min_not_us)
|
||||||
self.log.info(
|
# Check that this list without us is not empty
|
||||||
"Lowballing next competitor for {currency}: {margin}",
|
if ads_without_us:
|
||||||
currency=currency,
|
# Find the cheapest from these
|
||||||
margin=our_max_adjusted,
|
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)
|
||||||
return our_max_adjusted
|
# 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:
|
else:
|
||||||
our_max_adjusted = float(settings.Agora.MaxMargin)
|
self.log.debug("All ads are NOT ours for: {x}", x=currency)
|
||||||
# Set the max rates as people have no choice
|
# All the ads are not ours, but we are first...
|
||||||
return our_max_adjusted
|
# Check if the lowballed, lowest (that is not ours) ad's margin
|
||||||
iter_count = 0
|
# is less than our minimum
|
||||||
while place > 1 and not iter_count > 100:
|
if lowball_lowest_not_ours < float(settings.Agora.MinMargin):
|
||||||
iter_count += 1
|
self.log.debug("Lowball lowest not ours less than MinMargin")
|
||||||
recommended_margin = float(ads[place][3]) - 0.01
|
return float(settings.Agora.MinMargin)
|
||||||
if recommended_margin <= float(settings.Agora.MinMargin):
|
else:
|
||||||
break
|
self.log.debug("Returning lowballed figure: {x}", x=lowball_lowest_not_ours)
|
||||||
ads[place][3] = recommended_margin
|
return lowball_lowest_not_ours
|
||||||
print("Set recommended margin for", currency, recommended_margin)
|
else:
|
||||||
|
self.log.debug("We are NOT the cheapest for: {x}", x=currency)
|
||||||
ads, place = self.sort_ads_annotate_place(ads)
|
# We are not the cheapest :(
|
||||||
# Return our adjusted (or left) margin
|
# Check if this list is empty
|
||||||
return float(ads[place][3])
|
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):
|
def nuke_ads(self):
|
||||||
"""
|
"""
|
||||||
|
|
Loading…
Reference in New Issue