Begin adding platform support

app/local_settings.py

@@ -36,7 +36,8 @@ ELASTICSEARCH_USERNAME = getenv("ELASTICSEARCH_USERNAME", "elastic")
 ELASTICSEARCH_PASSWORD = getenv("ELASTICSEARCH_PASSWORD", "changeme")
 ELASTICSEARCH_HOST = getenv("ELASTICSEARCH_HOST", "localhost")
 ELASTICSEARCH_TLS = getenv("ELASTICSEARCH_TLS", "false") in trues
-
+ELASTICSEARCH_INDEX = getenv("ELASTICSEARCH_INDEX", "pluto")
+ELASTICSEARCH_INDEX_ADS = getenv("ELASTICSEARCH_INDEX_ADS", "ads")
 DEBUG = getenv("DEBUG", "false").lower() in trues
 PROFILER = getenv("PROFILER", "false").lower() in trues
 

app/urls.py

@@ -20,7 +20,7 @@ from django.contrib.auth.views import LogoutView
 from django.urls import include, path
 from two_factor.urls import urlpatterns as tf_urls
 
-from core.views import aggregators, banks, base, notifications
+from core.views import aggregators, banks, base, notifications, platforms
 
 # from core.views.stripe_callbacks import Callback
 
@@ -122,4 +122,25 @@ urlpatterns = [
         banks.BanksTransactions.as_view(),
         name="transactions",
     ),
+    # Platforms
+    path(
+        "platforms/<str:type>/",
+        platforms.PlatformList.as_view(),
+        name="platforms",
+    ),
+    path(
+        "platforms/<str:type>/create/",
+        platforms.PlatformCreate.as_view(),
+        name="platform_create",
+    ),
+    path(
+        "platforms/<str:type>/update/<str:pk>/",
+        platforms.PlatformUpdate.as_view(),
+        name="platform_update",
+    ),
+    path(
+        "platforms/<str:type>/delete/<str:pk>/",
+        platforms.PlatformDelete.as_view(),
+        name="platform_delete",
+    ),
 ] + static(settings.STATIC_URL, document_root=settings.STATIC_ROOT)

core/clients/aggregator.py

@@ -1,6 +1,11 @@
 from abc import ABC
 
-from core.lib.db import convert, r
+from core.lib import db, notify
+
+# from core.lib.money import money
+from core.util import logs
+
+log = logs.get_logger("aggregator")
 
 
 class AggregatorClient(ABC):
@@ -52,18 +57,312 @@ class AggregatorClient(ABC):
         # for transaction_id in transaction_ids:
         if not transaction_ids:
             return
-        r.sadd(new_key_name, *transaction_ids)
+        db.r.sadd(new_key_name, *transaction_ids)
 
-        difference = list(r.sdiff(new_key_name, old_key_name))
+        difference = list(db.r.sdiff(new_key_name, old_key_name))
 
-        difference = convert(difference)
+        difference = db.convert(difference)
 
         new_transactions = [
             x for x in transactions if x["transaction_id"] in difference
         ]
 
         # Rename the new key to the old key so we can run the diff again
-        r.rename(new_key_name, old_key_name)
+        db.r.rename(new_key_name, old_key_name)
         for transaction in new_transactions:
             transaction["subclass"] = self.name
             # self.tx.transaction(transaction)
+
+    def valid_transaction(self, data):
+        """
+        Determine if a given transaction object is valid.
+        :param data: a transaction cast
+        :type data: dict
+        :return: whether the transaction is valid
+        :rtype: bool
+        """
+        txid = data["transaction_id"]
+        if "amount" not in data:
+            return False
+        if "currency" not in data:
+            return False
+        amount = data["amount"]
+        if amount <= 0:
+            log.info(f"Ignoring transaction with negative/zero amount: {txid}")
+            return False
+        return True
+
+    def extract_reference(self, data):
+        """
+        Extract a reference from the transaction cast.
+        :param data: a transaction cast
+        :type data: dict
+        :return: the extracted reference or not_set
+        :rtype: str
+        """
+        if "reference" in data:
+            return data["reference"]
+        elif "meta" in data:
+            if "provider_reference" in data["meta"]:
+                return data["meta"]["provider_reference"]
+        return "not_set"
+
+    def extract_sender(self, data):
+        """
+        Extract a sender name from the transaction cast.
+        :param data: a transaction cast
+        :type data: dict
+        :return: the sender name or not_set
+        :rtype: str
+        """
+        if "debtorName" in data:
+            return data["debtorName"]
+        elif "meta" in data:
+            if "debtor_account_name" in data["meta"]:
+                return data["meta"]["debtor_account_name"]
+        elif "  " in data["reference"]:
+            refsplit = data["reference"].split("  ")
+            if not len(refsplit) == 2:
+                log.error(f"Sender cannot be extracted: {data}")
+                return "not_set"
+            realname, part2 = data["reference"].split("  ")
+            return realname
+
+        return "not_set"
+
+    async def reference_partial_check(self, reference, txid, currency, amount):
+        """
+        Perform a partial check by intersecting all parts of the split of the
+        reference against the existing references, and returning a set of the matches.
+        :param reference: the reference to check
+        :type reference: str
+        :return: matching trade ID string
+        :rtype: str
+        """
+        # Partial reference implementation
+        # Account for silly people not removing the default string
+        # Split the reference into parts
+        ref_split = reference.split(" ")
+        # Get all existing references
+        existing_refs = await db.get_refs()
+        # Get all parts of the given reference split that match the existing references
+        stored_trade_reference = set(existing_refs).intersection(set(ref_split))
+        if len(stored_trade_reference) > 1:
+            message = (
+                f"Multiple references valid for TXID {txid}: {reference}"
+                f"Currency: {currency} | Amount: {amount}"
+            )
+            title = "Error: multiple references valid"
+            await notify.sendmsg(self.instance.user, message, title=title)
+            return False
+        if len(stored_trade_reference) == 0:
+            return None
+        return stored_trade_reference.pop()
+
+    async def can_alt_lookup(self, amount, currency, reference):
+        amount_usd = self.money.to_usd(amount, currency)
+        # Amount is reliable here as it is checked by find_trade,
+        # so no need for stored_trade["amount"]
+        if float(amount_usd) > float(settings.Agora.AcceptableAltLookupUSD):
+            message = (
+                f"Amount exceeds max for {reference}"
+                f"Currency: {currency} | Amount: {amount}"
+            )
+            title = "Amount exceeds max for {reference}"
+            await notify.sendmsg(self.instance.user, message, title=title)
+            return False
+        return True
+
+    async def amount_currency_lookup(self, amount, currency, txid, reference):
+        log.info(f"No reference in DB refs for {reference}")
+        self.irc.sendmsg(f"No reference in DB refs for {reference}")
+        # Try checking just amount and currency, as some people
+        # (usually people buying small amounts)
+        #  are unable to put in a reference properly.
+
+        log.info(f"Checking against amount and currency for TXID {txid}")
+        self.irc.sendmsg(f"Checking against amount and currency for TXID {txid}")
+        title = f"Checking against amount and currency for TXID {txid}"
+        message = (
+            f"Checking against amount and currency for TXID {txid}"
+            f"Currency: {currency} | Amount: {amount}"
+        )
+        await notify.sendmsg(self.instance.user, message, title=title)
+        if not await self.can_alt_lookup(amount, currency, reference):
+            return False
+        stored_trade = await self.find_trade(txid, currency, amount)
+        if not stored_trade:
+            title = f"Failed to get reference by amount and currency: {txid}"
+            message = (
+                f"Failed to get reference by amount and currency: {txid}"
+                f"Currency: {currency} | Amount: {amount}"
+            )
+            await notify.sendmsg(self.instance.user, message, title=title)
+            return None
+        stored_trade["amount"] = float(stored_trade["amount"])  # convert to float
+        return stored_trade
+
+    async def normal_lookup(self, stored_trade_reference, reference, currency, amount):
+        stored_trade = await db.get_ref(stored_trade_reference)
+        if not stored_trade:
+            title = f"No reference in DB for {reference}"
+            message = (
+                f"No reference in DB for {reference}"
+                f"Currency: {currency} | Amount: {amount}"
+            )
+            await notify.sendmsg(self.instance.user, message, title=title)
+            return False
+        stored_trade["amount"] = float(stored_trade["amount"])  # convert to float
+        return stored_trade
+
+    async def currency_check(self, currency, amount, reference, stored_trade):
+        if not stored_trade["currency"] == currency:
+            title = "Currency mismatch"
+            message = (
+                f"Currency mismatch, Agora: {stored_trade['currency']} "
+                f"/ Sink: {currency}"
+            )
+            await notify.sendmsg(self.instance.user, message, title=title)
+            return False
+        return True
+
+    async def alt_amount_check(
+        self, platform, amount, currency, reference, stored_trade
+    ):
+        # If the amount does not match exactly, get the min and max values for our
+        # given acceptable margins for trades
+        min_amount, max_amount = await self.money.get_acceptable_margins(
+            platform, currency, stored_trade["amount"]
+        )
+        log.info(
+            (
+                f"Amount does not match exactly, trying with margins: min: {min_amount}"
+                f" / max: {max_amount}"
+            )
+        )
+        self.irc.sendmsg()
+        title = "Amount does not match exactly"
+        message = (
+            f"Amount does not match exactly, trying with margins: min: "
+            f"{min_amount} / max: {max_amount}"
+        )
+        await notify.sendmsg(self.instance.user, message, title=title)
+        if not min_amount < amount < max_amount:
+            title = "Amount mismatch - not in margins"
+            message = (
+                f"Amount mismatch - not in margins: {stored_trade['amount']} "
+                f"(min: {min_amount} / max: {max_amount}"
+            )
+            await notify.sendmsg(self.instance.user, message, title=title)
+            return False
+        return True
+
+    async def transaction(self, data):
+        """
+        Store details of transaction and post notifications to IRC.
+        Matches it up with data stored in Redis to attempt to reconcile with an Agora
+        trade.
+        :param data: details of transaction
+        :type data: dict
+        """
+        valid = self.valid_transaction(data)
+        if not valid:
+            return False
+        ts = data["timestamp"]
+        txid = data["transaction_id"]
+        amount = float(data["amount"])
+        currency = data["currency"]
+
+        reference = self.extract_reference(data)
+        sender = self.extract_sender(data)
+
+        subclass = data["subclass"]
+        to_store = {
+            "subclass": subclass,
+            "ts": ts,
+            "txid": txid,
+            "reference": reference,
+            "amount": amount,
+            "currency": currency,
+            "sender": sender,
+        }
+        db.r.hmset(f"tx.{txid}", to_store)
+
+        log.info(f"Transaction processed: {dumps(to_store, indent=2)}")
+        self.irc.sendmsg(
+            (
+                f"AUTO Incoming transaction on {subclass}: {txid} {amount}{currency} "
+                f"({reference})"
+            )
+        )
+
+        stored_trade_reference = self.reference_partial_check(
+            reference, txid, currency, amount
+        )
+        if stored_trade_reference is False:  # can be None though
+            return
+
+        stored_trade = False
+        looked_up_without_reference = False
+
+        # Normal implementation for when we have a reference
+        if stored_trade_reference:
+            stored_trade = self.normal_lookup(
+                stored_trade_reference, reference, currency, amount
+            )
+            # if not stored_trade:
+            #    return
+
+        # Amount/currency lookup implementation for when we have no reference
+        else:
+            if not stored_trade:  # check we don't overwrite the lookup above
+                stored_trade = self.amount_currency_lookup(
+                    amount, currency, txid, reference
+                )
+                if stored_trade is False:
+                    return
+                if stored_trade:
+                    # Note that we have looked it up without reference so we don't use
+                    # +- below
+                    # This might be redundant given the amount checks in find_trade,
+                    # but better safe than sorry!
+                    looked_up_without_reference = True
+                else:
+                    return
+            else:
+                # Stored trade reference is none, the checks below will do nothing
+                return
+
+        # Make sure it was sent in the expected currency
+        if not self.currency_check(currency, amount, reference, stored_trade):
+            return
+
+        # Make sure the expected amount was sent
+        if not stored_trade["amount"] == amount:
+            if looked_up_without_reference:
+                return
+            platform = stored_trade["subclass"]
+            if not self.alt_amount_check(
+                platform, amount, currency, reference, stored_trade
+            ):
+                return
+        platform = stored_trade["subclass"]
+        platform_buyer = stored_trade["buyer"]
+
+        # Check sender - we don't do anything with this yet
+        # sender_valid = self.antifraud.check_valid_sender(
+        #     reference, platform, sender, platform_buyer
+        # )
+        # log.info(f"Trade {reference} buyer {platform_buyer} valid: {sender_valid}")
+        # trade_released = self.release_map_trade(reference, txid)
+        # if trade_released:
+        #    self.ux.notify.notify_complete_trade(amount, currency)
+        # else:
+        #    log.error(f"Cannot release trade {reference}.")
+        #    return
+
+        rtrn = await self.release_funds(stored_trade["id"], stored_trade["reference"])
+        if rtrn:
+            title = "Trade complete"
+            message = f"Trade complete: {amount}{currency}"
+            await notify.sendmsg(self.instance.user, message, title=title)

core/clients/platforms/agora.py

@@ -0,0 +1,120 @@
+# Other library imports
+from pyotp import TOTP
+
+from core.clients.base import BaseClient
+from core.clients.platform import LocalPlatformClient
+from core.lib.money import Money
+
+
+class AgoraClient(BaseClient, LocalPlatformClient):
+    """
+    AgoraDesk API handler.
+    """
+
+    async 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
+        """
+        print("CALLING RELEASE FUNDS", contact_id)
+        if self.instance.dummy:
+            self.log.error(
+                f"Running in dummy mode, not releasing funds for {contact_id}"
+            )
+            return
+        payload = {"tradeId": contact_id, "password": self.sets.Pass}
+        rtrn = await self.api._api_call(
+            api_method=f"contact_release/{contact_id}",
+            http_method="POST",
+            query_values=payload,
+        )
+
+        # Check if we can withdraw funds
+        await self.withdraw_funds()
+
+        return rtrn
+
+    # TODO: write test before re-enabling adding total_trades
+    async def withdraw_funds(self):
+        """
+        Withdraw excess funds to our XMR wallets.
+        """
+        print("CALLING WITHDRAW FUNDS")
+        totals_all = await self.money.get_total()
+        if totals_all is False:
+            return False
+
+        wallet_xmr, _ = totals_all[2]
+
+        # Get the wallet balances in USD
+        total_usd = totals_all[0][1]
+
+        # total_trades_usd = self.tx.get_open_trades_usd()
+        if not total_usd:
+            return False
+        # total_usd += total_trades_usd
+
+        profit_usd = total_usd - float(settings.Money.BaseUSD)
+        # Get the XMR -> USD exchange rate
+        xmr_usd = self.money.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(
+                (
+                    f"Not enough funds to withdraw {profit_usd_in_xmr}, "
+                    f"as wallet only contains {wallet_xmr}"
+                )
+            )
+            self.irc.sendmsg(
+                (
+                    f"Not enough funds to withdraw {profit_usd_in_xmr}, "
+                    f"as wallet only contains {wallet_xmr}"
+                )
+            )
+            self.ux.notify.notify_need_topup(profit_usd_in_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)
+
+        # Read OTP secret
+        with open("otp.key", "r") as f:
+            otp_key = f.read()
+        f.close()
+        otp_key = otp_key.replace("\n", "")
+
+        # Get OTP code
+        otp_code = TOTP(otp_key)
+
+        # Set up the format for calling wallet_send_xmr
+        send_cast = {
+            "address": None,
+            "amount": half_rounded,
+            "password": settings.Agora.Pass,
+            "otp": otp_code.now(),
+        }
+
+        send_cast["address"] = settings.XMR.Wallet1
+        rtrn1 = await self.api.wallet_send_xmr(**send_cast)
+
+        send_cast["address"] = settings.XMR.Wallet2
+        rtrn2 = await self.api.wallet_send_xmr(**send_cast)
+
+        self.irc.sendmsg(f"Withdrawal: {rtrn1['success']} | {rtrn2['success']}")
+        self.ux.notify.notify_withdrawal(half_rounded)

core/forms.py

@@ -4,7 +4,7 @@ from django.core.exceptions import FieldDoesNotExist
 from django.forms import ModelForm
 from mixins.restrictions import RestrictedFormMixin
 
-from .models import Aggregator, NotificationSettings, User
+from .models import Aggregator, NotificationSettings, Platform, User
 
 # flake8: noqa: E501
 
@@ -73,3 +73,58 @@ class AggregatorForm(RestrictedFormMixin, ModelForm):
             "poll_interval": "The interval in seconds to poll the aggregator service.",
             "enabled": "Whether or not the aggregator connection is enabled.",
         }
+
+
+class PlatformForm(RestrictedFormMixin, ModelForm):
+    def __init__(self, *args, **kwargs):
+        super(PlatformForm, self).__init__(*args, **kwargs)
+        upper = ["usd", "otp"]
+        for field in self.fields:
+            for up in upper:
+                if up in self.fields[field].label:
+                    self.fields[field].label = self.fields[field].label.replace(
+                        up, up.upper()
+                    )
+
+    class Meta:
+        model = Platform
+        fields = (
+            "name",
+            "service",
+            "token",
+            "password",
+            "otp_token",
+            "username",
+            "send",
+            "cheat",
+            "dummy",
+            "cheat_interval_seconds",
+            "margin",
+            "max_margin",
+            "min_margin",
+            "min_trade_size_usd",
+            "max_trade_size_usd",
+            "accept_within_usd",
+            "no_reference_amount_check_max_usd",
+            "enabled",
+        )
+        help_texts = {
+            "name": "The name of the platform connection.",
+            "service": "The platform service to use.",
+            "token": "The JWT auth token.",
+            "password": "Account password",
+            "otp_token": "The OTP secret key.",
+            "username": "Account username",
+            "send": "Whether or not to send messages on new trades.",
+            "cheat": "Whether or not to run the Autoprice cheat.",
+            "dummy": "When enabled, the trade escrow feature will be disabled.",
+            "cheat_interval_seconds": "The interval in seconds to run the Autoprice cheat.",
+            "margin": "The current margin. Only valid for initial ads post. Autoprice will override this.",
+            "max_margin": "The maximum margin to use.",
+            "min_margin": "The minimum margin to use.",
+            "min_trade_size_usd": "The minimum trade size in USD.",
+            "max_trade_size_usd": "The maximum trade size in USD.",
+            "accept_within_usd": "When a trade is wrong by less than this amount, it will be accepted.",
+            "no_reference_amount_check_max_usd": "When ticked, when no reference was found and a trade is higher than this amount, we will not accept payment even if it is the only one with this amount.",
+            "enabled": "Whether or not the platform connection is enabled.",
+        }

core/lib/antifraud.py

@@ -0,0 +1,111 @@
+# Project imports
+from core.lib import db, notify
+from core.util import logs
+
+log = logs.get_logger("antifraud")
+
+
+class AntiFraud(object):
+    async def add_bank_sender(self, platform, platform_buyer, bank_sender):
+        """
+        Add the bank senders into Redis.
+        :param platform: name of the platform - freeform
+        :param platform_buyer: the username of the buyer on the platform
+        :param bank_sender: the sender name from the bank
+        """
+        key = f"namemap.{platform}.{platform_buyer}"
+        await db.r.sadd(key, bank_sender)
+
+    async def get_previous_senders(self, platform, platform_buyer):
+        """
+        Get all the previous bank sender names for the given buyer on the platform.
+        :param platform: name of the platform - freeform
+        :param platform_buyer: the username of the buyer on the platform
+        :return: set of previous buyers
+        :rtype: set
+        """
+        key = f"namemap.{platform}.{platform_buyer}"
+        senders = await db.r.smembers(key)
+        if not senders:
+            return None
+        senders = db.convert(senders)
+        return senders
+
+    async def check_valid_sender(
+        self, reference, platform, bank_sender, platform_buyer
+    ):
+        """
+        Check that either:
+        * The platform buyer has never had a recognised transaction before
+        * The bank sender name matches a previous transaction from the platform buyer
+        :param reference: the trade reference
+        :param platform: name of the platform - freeform
+        :param bank_sender: the sender of the bank transaction
+        :param platform_buyer: the username of the buyer on the platform
+        :return: whether the sender is valid
+        :rtype: bool
+        """
+        senders = await self.get_previous_senders(platform, platform_buyer)
+        if senders is None:  # no senders yet, assume it's valid
+            return True
+        if platform_buyer in senders:
+            return True
+        self.ux.notify.notify_sender_name_mismatch(
+            reference, platform_buyer, bank_sender
+        )
+        title = "Sender name mismatch"
+        message = (
+            f"Sender name mismatch for {reference}:\n"
+            f"Platform buyer: {platform_buyer}"
+            f"Bank sender: {bank_sender}"
+        )
+        # await notify.sendmsg(self.instance.) # TODO
+        return False
+
+    async def check_tx_sender(self, tx, reference):
+        """
+        Check whether the sender of a given transaction is authorised based on the previous
+        transactions of the username that originated the trade reference.
+        :param tx: the transaction ID
+        :param reference: the trade reference
+        """
+        stored_trade = await db.get_ref(reference)
+        if not stored_trade:
+            return None
+        stored_tx = await db.get_tx(tx)
+        if not stored_tx:
+            return None
+        bank_sender = stored_tx["sender"]
+        platform_buyer = stored_trade["buyer"]
+        platform = stored_trade["subclass"]
+        is_allowed = await self.check_valid_sender(
+            reference, platform, bank_sender, platform_buyer
+        )
+        if is_allowed is True:
+            return True
+        return False
+
+    # def user_verification_successful(self, uid):
+    #     """
+    #     A user has successfully completed verification.
+    #     """
+    #     self.log.info(f"User has completed verification: {uid}")
+    #     trade_list = self.markets.find_trades_by_uid(uid)
+    #     for platform, trade_id, reference, currency in trade_list:
+    #         self.markets.send_bank_details(platform, currency, trade_id)
+    #         self.markets.send_reference(platform, trade_id, reference)
+
+    # def send_verification_url(self, platform, uid, trade_id):
+    #     send_setting, post_message = self.markets.get_send_settings(platform)
+    #     if send_setting == "1":
+    #         auth_url = self.ux.verify.create_applicant_and_get_link(uid)
+    #         if platform == "lbtc":
+    #             auth_url = auth_url.replace("https://", "")  # hack
+    #         post_message(
+    #             trade_id,
+    #             f"Hi! To continue the trade, please complete the verification form: {auth_url}",
+    #         )
+
+
+if __name__ == "__main__":
+    antifraud = AntiFraud()

core/lib/db.py

@@ -2,7 +2,7 @@ from redis import asyncio as aioredis
 
 from core.util import logs
 
-log = logs.get_logger("scheduling")
+log = logs.get_logger("db")
 
 r = aioredis.from_url("redis://redis:6379", db=0)
 
@@ -22,20 +22,20 @@ def convert(data):
     return data
 
 
-def get_refs():
+async def get_refs():
     """
     Get all reference IDs for trades.
     :return: list of trade IDs
     :rtype: list
     """
     references = []
-    ref_keys = r.keys("trade.*.reference")
+    ref_keys = await r.keys("trade.*.reference")
     for key in ref_keys:
         references.append(r.get(key))
     return convert(references)
 
 
-def tx_to_ref(tx):
+async def tx_to_ref(tx):
     """
     Convert a trade ID to a reference.
     :param tx: trade ID
@@ -43,16 +43,16 @@ def tx_to_ref(tx):
     :return: reference
     :rtype: string
     """
-    refs = get_refs()
+    refs = await get_refs()
     for reference in refs:
-        ref_data = convert(r.hgetall(f"trade.{reference}"))
+        ref_data = convert(await r.hgetall(f"trade.{reference}"))
         if not ref_data:
             continue
         if ref_data["id"] == tx:
             return reference
 
 
-def ref_to_tx(reference):
+async def ref_to_tx(reference):
     """
     Convert a reference to a trade ID.
     :param reference: trade reference
@@ -60,27 +60,27 @@ def ref_to_tx(reference):
     :return: trade ID
     :rtype: string
     """
-    ref_data = convert(r.hgetall(f"trade.{reference}"))
+    ref_data = convert(await r.hgetall(f"trade.{reference}"))
     if not ref_data:
         return False
     return ref_data["id"]
 
 
-def get_ref_map():
+async def get_ref_map():
     """
     Get all reference IDs for trades.
     :return: dict of references keyed by TXID
     :rtype: dict
     """
     references = {}
-    ref_keys = r.keys("trade.*.reference")
+    ref_keys = await r.keys("trade.*.reference")
     for key in ref_keys:
         tx = convert(key).split(".")[1]
-        references[tx] = r.get(key)
+        references[tx] = await r.get(key)
     return convert(references)
 
 
-def get_ref(reference):
+async def get_ref(reference):
     """
     Get the trade information for a reference.
     :param reference: trade reference
@@ -88,7 +88,7 @@ def get_ref(reference):
     :return: dict of trade information
     :rtype: dict
     """
-    ref_data = r.hgetall(f"trade.{reference}")
+    ref_data = await r.hgetall(f"trade.{reference}")
     ref_data = convert(ref_data)
     if "subclass" not in ref_data:
         ref_data["subclass"] = "agora"
@@ -97,7 +97,7 @@ def get_ref(reference):
     return ref_data
 
 
-def get_tx(tx):
+async def get_tx(tx):
     """
     Get the transaction information for a transaction ID.
     :param reference: trade reference
@@ -105,31 +105,31 @@ def get_tx(tx):
     :return: dict of trade information
     :rtype: dict
     """
-    tx_data = r.hgetall(f"tx.{tx}")
+    tx_data = await r.hgetall(f"tx.{tx}")
     tx_data = convert(tx_data)
     if not tx_data:
         return False
     return tx_data
 
 
-def get_subclass(reference):
+async def get_subclass(reference):
-    obj = r.hget(f"trade.{reference}", "subclass")
+    obj = await r.hget(f"trade.{reference}", "subclass")
     subclass = convert(obj)
     return subclass
 
 
-def del_ref(reference):
+async def del_ref(reference):
     """
     Delete a given reference from the Redis database.
     :param reference: trade reference to delete
     :type reference: string
     """
-    tx = ref_to_tx(reference)
+    tx = await ref_to_tx(reference)
-    r.delete(f"trade.{reference}")
+    await r.delete(f"trade.{reference}")
-    r.delete(f"trade.{tx}.reference")
+    await r.delete(f"trade.{tx}.reference")
 
 
-def cleanup(subclass, references):
+async def cleanup(subclass, references):
     """
     Reconcile the internal reference database with a given list of references.
     Delete all internal references not present in the list and clean up artifacts.
@@ -137,14 +137,44 @@ def cleanup(subclass, references):
     :type references: list
     """
     messages = []
-    for tx, reference in get_ref_map().items():
+    for tx, reference in await get_ref_map().items():
         if reference not in references:
-            if get_subclass(reference) == subclass:
+            if await get_subclass(reference) == subclass:
                 logmessage = (
                     f"[{reference}] ({subclass}): Archiving trade reference. TX: {tx}"
                 )
                 messages.append(logmessage)
                 log.info(logmessage)
-                r.rename(f"trade.{tx}.reference", f"archive.trade.{tx}.reference")
+                await r.rename(f"trade.{tx}.reference", f"archive.trade.{tx}.reference")
-                r.rename(f"trade.{reference}", f"archive.trade.{reference}")
+                await r.rename(f"trade.{reference}", f"archive.trade.{reference}")
     return messages
+
+
+async def find_trade(self, txid, currency, amount):
+    """
+    Get a trade reference that matches the given currency and amount.
+    Only works if there is one result.
+    :param txid: Sink transaction ID
+    :param currency: currency
+    :param amount: amount
+    :type txid: string
+    :type currency: string
+    :type amount: int
+    :return: matching trade object or False
+    :rtype: dict or bool
+    """
+    refs = await get_refs()
+    matching_refs = []
+    # TODO: use get_ref_map in this function instead of calling get_ref multiple times
+    for ref in refs:
+        stored_trade = await get_ref(ref)
+        if stored_trade["currency"] == currency and float(
+            stored_trade["amount"]
+        ) == float(amount):
+            matching_refs.append(stored_trade)
+    if len(matching_refs) != 1:
+        log.error(
+            f"Find trade returned multiple results for TXID {txid}: {matching_refs}"
+        )
+        return False
+    return matching_refs[0]

core/lib/money.py

@@ -0,0 +1,497 @@
+# Twisted imports
+import logging
+from datetime import datetime
+
+import urllib3
+
+# Other library imports
+from aiocoingecko import AsyncCoinGeckoAPISession
+from django.conf import settings
+from elasticsearch import AsyncElasticsearch
+from forex_python.converter import CurrencyRates
+
+# TODO: secure ES traffic properly
+urllib3.disable_warnings()
+
+tracer = logging.getLogger("opensearch")
+tracer.setLevel(logging.CRITICAL)
+tracer = logging.getLogger("elastic_transport.transport")
+tracer.setLevel(logging.CRITICAL)
+
+
+class Money(object):
+    """
+    Generic class for handling money-related matters that aren't Revolut or Agora.
+    """
+
+    def __init__(self):
+        """
+        Initialise the Money object.
+        Set the logger.
+        Initialise the CoinGecko API.
+        """
+        print("MONEY INIT")
+        self.cr = CurrencyRates()
+        self.cg = AsyncCoinGeckoAPISession()
+        auth = (settings.ELASTICSEARCH_USERNAME, settings.ELASTICSEARCH_PASSWORD)
+        client = AsyncElasticsearch(
+            settings.ELASTICSEARCH_HOST, http_auth=auth, verify_certs=False
+        )
+        self.es = client
+
+    async def run_checks_in_thread(self):
+        """
+        Run all the balance checks that output into ES in another thread.
+        """
+        total = await self.get_total()
+        remaining = await self.get_remaining()
+        profit = await self.get_profit()
+        profit_with_trades = await self.get_profit(True)
+        open_trades = await self.get_open_trades_usd()
+        total_remaining = await self.get_total_remaining()
+        total_with_trades = await self.get_total_with_trades()
+        # This will make them all run concurrently, hopefully not hitting rate limits
+        for x in (
+            total,
+            remaining,
+            profit,
+            profit_with_trades,
+            open_trades,
+            total_remaining,
+            total_with_trades,
+        ):
+            yield x
+
+    # def setup_loops(self):
+    #     """
+    #     Set up the LoopingCalls to get the balance so we have data in ES.
+    #     """
+    #     if settings.ES.Enabled == "1" or settings.Logstash.Enabled == "1":
+    #         self.lc_es_checks = LoopingCall(self.run_checks_in_thread)
+    #         delay = int(settings.ES.RefreshSec)
+    #         self.lc_es_checks.start(delay)
+    #         if settings.ES.Enabled == "1":
+    #             self.agora.es = self.es
+    #             self.lbtc.es = self.es
+
+    async def write_to_es(self, msgtype, cast):
+        cast["type"] = "money"
+        cast["ts"] = str(datetime.now().isoformat())
+        cast["xtype"] = msgtype
+        cast["user_id"] = self.instance.user.id
+        cast["platform_id"] = self.instance.id
+        await self.es.index(index=settings.ELASTICSEARCH_INDEX, body=cast)
+
+    async def lookup_rates(self, platform, ads, rates=None):
+        """
+        Lookup the rates for a list of public ads.
+        """
+        if not rates:
+            rates = await self.cg.get_price(
+                ids=["monero", "bitcoin"],
+                vs_currencies=self.markets.get_all_currencies(platform),
+            )
+            # Set the price based on the asset
+        for ad in ads:
+            if ad[4] == "XMR":
+                coin = "monero"
+            elif ad[4] == "BTC":
+                coin = "bitcoin"  # No s here
+            currency = ad[5]
+            base_currency_price = rates[coin][currency.lower()]
+            price = float(ad[2])
+            rate = round(price / base_currency_price, 2)
+            ad.append(rate)
+        # TODO: sort?
+        return sorted(ads, key=lambda x: x[2])
+
+    async def get_rates_all(self):
+        """
+        Get all rates that pair with USD.
+        :return: dictionary of USD/XXX rates
+        :rtype: dict
+        """
+        rates = await self.cr.get_rates("USD")
+        return rates
+
+    async def get_acceptable_margins(self, platform, 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
+        """
+        sets = util.get_settings(platform)
+        rates = await self.get_rates_all()
+        if currency == "USD":
+            min_amount = amount - float(sets.AcceptableUSDMargin)
+            max_amount = amount + float(sets.AcceptableUSDMargin)
+            return (min_amount, max_amount)
+        amount_usd = amount / rates[currency]
+        min_usd = amount_usd - float(sets.AcceptableUSDMargin)
+        max_usd = amount_usd + float(sets.AcceptableUSDMargin)
+        min_local = min_usd * rates[currency]
+        max_local = max_usd * rates[currency]
+        return (min_local, max_local)
+
+    async def get_minmax(self, platform, asset, currency):
+        sets = util.get_settings(platform)
+        rates = await self.get_rates_all()
+        if currency not in rates and not currency == "USD":
+            self.log.error(f"Can't create ad without rates: {currency}")
+            return
+        if asset == "XMR":
+            min_usd = float(sets.MinUSDXMR)
+            max_usd = float(sets.MaxUSDXMR)
+        elif asset == "BTC":
+            min_usd = float(sets.MinUSDBTC)
+            max_usd = float(sets.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
+
+        return (min_amount, max_amount)
+
+    async def to_usd(self, amount, currency):
+        if currency == "USD":
+            return float(amount)
+        else:
+            rates = await self.get_rates_all()
+            return float(amount) / rates[currency]
+
+    async def multiple_to_usd(self, currency_map):
+        """
+        Convert multiple curencies to USD while saving API calls.
+        """
+        rates = await self.get_rates_all()
+        cumul = 0
+        for currency, amount in currency_map.items():
+            if currency == "USD":
+                cumul += float(amount)
+            else:
+                cumul += float(amount) / rates[currency]
+        return cumul
+
+    async def get_profit(self, trades=False):
+        """
+        Check how much total profit we have made.
+        :return: profit in USD
+        :rtype: float
+        """
+        total_usd = await self.get_total_usd()
+        if not total_usd:
+            return False
+        if trades:
+            trades_usd = await self.get_open_trades_usd()
+            total_usd += trades_usd
+
+        profit = total_usd - float(settings.Money.BaseUSD)
+        if trades:
+            cast_es = {
+                "profit_trades_usd": profit,
+            }
+        else:
+            cast_es = {
+                "profit_usd": profit,
+            }
+
+        await self.write_to_es("get_profit", cast_es)
+        return profit
+
+    async def get_total_usd(self):
+        """
+        Get total USD in all our accounts, bank and trading.
+        :return: value in USD
+        :rtype float:
+        """
+        total_sinks_usd = await self.sinks.get_total_usd()
+        agora_wallet_xmr = await self.agora.api.wallet_balance_xmr()
+        agora_wallet_btc = await self.agora.api.wallet_balance()
+        # lbtc_wallet_btc = await self.lbtc.api.wallet_balance()
+        if not agora_wallet_xmr["success"]:
+            return False
+        if not agora_wallet_btc["success"]:
+            return False
+        # if not lbtc_wallet_btc["success"]:
+        #     return False
+        if not agora_wallet_xmr["response"]:
+            return False
+        if not agora_wallet_btc["response"]:
+            return False
+        # if not lbtc_wallet_btc["response"]:
+        #     return False
+        total_xmr_agora = agora_wallet_xmr["response"]["data"]["total"]["balance"]
+        total_btc_agora = agora_wallet_btc["response"]["data"]["total"]["balance"]
+        # total_btc_lbtc = lbtc_wallet_btc["response"]["data"]["total"]["balance"]
+        # Get the XMR -> USD exchange rate
+        xmr_usd = await self.cg.get_price(ids="monero", vs_currencies=["USD"])
+
+        # Get the BTC -> USD exchange rate
+        btc_usd = await self.cg.get_price(ids="bitcoin", vs_currencies=["USD"])
+
+        # Convert the Agora BTC total to USD
+        total_usd_agora_btc = float(total_btc_agora) * btc_usd["bitcoin"]["usd"]
+
+        # Convert the LBTC BTC total to USD
+        # total_usd_lbtc_btc = float(total_btc_lbtc) * btc_usd["bitcoin"]["usd"]
+
+        # Convert the Agora XMR total to USD
+        total_usd_agora_xmr = float(total_xmr_agora) * xmr_usd["monero"]["usd"]
+
+        # Add it all up
+        total_usd_agora = total_usd_agora_xmr + total_usd_agora_btc
+        # total_usd_lbtc = total_usd_lbtc_btc
+        total_usd = total_usd_agora + total_sinks_usd
+        # total_usd_lbtc +
+        cast_es = {
+            "price_usd": total_usd,
+            "total_usd_agora_xmr": total_usd_agora_xmr,
+            "total_usd_agora_btc": total_usd_agora_btc,
+            # "total_usd_lbtc_btc": total_usd_lbtc_btc,
+            "total_xmr_agora": total_xmr_agora,
+            "total_btc_agora": total_btc_agora,
+            # "total_btc_lbtc": total_btc_lbtc,
+            "xmr_usd": xmr_usd["monero"]["usd"],
+            "btc_usd": btc_usd["bitcoin"]["usd"],
+            "total_sinks_usd": total_sinks_usd,
+            "total_usd_agora": total_usd_agora,
+        }
+        await self.write_to_es("get_total_usd", cast_es)
+        return total_usd
+
+    # TODO: possibly refactor this into smaller functions which don't return as much
+    # check if this is all really needed in the corresponding withdraw function
+    async def get_total(self):
+        """
+        Get all the values corresponding to the amount of money we hold.
+        :return: ((total SEK, total USD, total GBP),
+            (total XMR USD, total BTC USD),
+            (total XMR, total BTC))
+        :rtype: tuple(tuple(float, float, float),
+            tuple(float, float),
+            tuple(float, float))
+        """
+        total_sinks_usd = await self.sinks.get_total_usd()
+        agora_wallet_xmr = await self.agora.api.wallet_balance_xmr()
+        agora_wallet_btc = await self.agora.api.wallet_balance()
+        # lbtc_wallet_btc = await self.lbtc.api.wallet_balance()
+        if not agora_wallet_xmr["success"]:
+            return False
+        if not agora_wallet_btc["success"]:
+            return False
+        # if not lbtc_wallet_btc["success"]:
+        #     return False
+        if not agora_wallet_xmr["response"]:
+            return False
+        if not agora_wallet_btc["response"]:
+            return False
+        # if not lbtc_wallet_btc["response"]:
+        #     return False
+        total_xmr_agora = agora_wallet_xmr["response"]["data"]["total"]["balance"]
+        total_btc_agora = agora_wallet_btc["response"]["data"]["total"]["balance"]
+        # total_btc_lbtc = lbtc_wallet_btc["response"]["data"]["total"]["balance"]
+        # Get the XMR -> USD exchange rate
+        xmr_usd = self.cg.get_price(ids="monero", vs_currencies=["USD"])
+
+        # Get the BTC -> USD exchange rate
+        btc_usd = self.cg.get_price(ids="bitcoin", vs_currencies=["USD"])
+
+        # Convert the Agora XMR total to USD
+        total_usd_agora_xmr = float(total_xmr_agora) * xmr_usd["monero"]["usd"]
+
+        # Convert the Agora BTC total to USD
+        total_usd_agora_btc = float(total_btc_agora) * btc_usd["bitcoin"]["usd"]
+
+        # Convert the LBTC BTC total to USD
+        # total_usd_lbtc_btc = float(total_btc_lbtc) * btc_usd["bitcoin"]["usd"]
+
+        # Add it all up
+        total_usd_agora = total_usd_agora_xmr + total_usd_agora_btc
+        # total_usd_lbtc = total_usd_lbtc_btc
+        total_usd = total_usd_agora + total_sinks_usd
+        # total_usd_lbtc
+
+        total_btc_usd = total_usd_agora_btc  # + total_usd_lbtc_btc
+        total_xmr_usd = total_usd_agora_xmr
+
+        total_xmr = total_xmr_agora
+        total_btc = total_btc_agora
+        # total_btc_lbtc
+
+        # Convert the total USD price to GBP and SEK
+        rates = await self.get_rates_all()
+        price_sek = rates["SEK"] * total_usd
+        price_usd = total_usd
+        price_gbp = rates["GBP"] * total_usd
+
+        cast = (
+            (
+                price_sek,
+                price_usd,
+                price_gbp,
+            ),  # Total prices in our 3 favourite currencies
+            (
+                total_xmr_usd,
+                total_btc_usd,
+            ),  # Total USD balance in only Agora
+            (total_xmr, total_btc),
+        )  # Total XMR and BTC balance in Agora
+
+        cast_es = {
+            "price_sek": price_sek,
+            "price_usd": price_usd,
+            "price_gbp": price_gbp,
+            "total_usd_agora_xmr": total_usd_agora_xmr,
+            "total_usd_agora_btc": total_usd_agora_btc,
+            # "total_usd_lbtc_btc": total_usd_lbtc_btc,
+            "total_xmr_agora": total_xmr_agora,
+            "total_btc_agora": total_btc_agora,
+            # "total_btc_lbtc": total_btc_lbtc,
+            "xmr_usd": xmr_usd["monero"]["usd"],
+            "btc_usd": btc_usd["bitcoin"]["usd"],
+            "total_sinks_usd": total_sinks_usd,
+            "total_usd_agora": total_usd_agora,
+        }
+        await self.write_to_es("get_total", cast_es)
+        return cast
+
+    async def get_remaining(self):
+        """
+        Check how much profit we need to make in order to withdraw.
+        :return: profit remaining in USD
+        :rtype: float
+        """
+        total_usd = await self.get_total_usd()
+        if not total_usd:
+            return False
+
+        withdraw_threshold = float(settings.Money.BaseUSD) + float(
+            settings.Money.WithdrawLimit
+        )
+        remaining = withdraw_threshold - total_usd
+        cast_es = {
+            "remaining_usd": remaining,
+        }
+        await self.write_to_es("get_remaining", cast_es)
+        return remaining
+
+    async def open_trades_usd_parse_dash(self, platform, dash, rates):
+        cumul_usd = 0
+        for contact_id, contact in dash.items():
+            # We need created at in order to look up the historical prices
+            created_at = contact["data"]["created_at"]
+
+            # Reformat the date how CoinGecko likes
+            # 2022-05-02T11:17:14+00:00
+            if "+" in created_at:
+                date_split = created_at.split("+")
+                date_split[1].replace(".", "")
+                date_split[1].replace(":", "")
+                created_at = "+".join(date_split)
+                date_parsed = datetime.strptime(created_at, "%Y-%m-%dT%H:%M:%S%z")
+            else:
+                date_parsed = datetime.strptime(created_at, "%Y-%m-%dT%H:%M:%S.%fZ")
+
+            date_formatted = date_parsed.strftime("%d-%m-%Y")
+
+            # Get the historical rates for the right asset, extract the price
+            if platform == "agora":
+                asset = contact["data"]["advertisement"]["asset"]
+            elif platform == "lbtc":
+                asset = "BTC"
+            if asset == "XMR":
+                amount_crypto = contact["data"]["amount_xmr"]
+                history = await self.cg.get_coin_history_by_id(
+                    id="monero", date=date_formatted
+                )
+                if "market_data" not in history:
+                    return False
+                crypto_usd = float(history["market_data"]["current_price"]["usd"])
+            elif asset == "BTC":
+                amount_crypto = contact["data"]["amount_btc"]
+                history = await self.cg.get_coin_history_by_id(
+                    id="bitcoin", date=date_formatted
+                )
+                crypto_usd = float(history["market_data"]["current_price"]["usd"])
+            # Convert crypto to fiat
+            amount = float(amount_crypto) * crypto_usd
+            currency = contact["data"]["currency"]
+            if not contact["data"]["is_selling"]:
+                continue
+            if currency == "USD":
+                cumul_usd += float(amount)
+            else:
+                rate = rates[currency]
+                amount_usd = float(amount) / rate
+                cumul_usd += amount_usd
+        return cumul_usd
+
+    async def get_open_trades_usd(self):
+        """
+        Get total value of open trades in USD.
+        :return: total trade value
+        :rtype: float
+        """
+        dash_agora = await self.agora.wrap_dashboard()
+        # dash_lbtc = self.lbtc.wrap_dashboard()
+        # dash_lbtc = yield dash_lbtc
+        if dash_agora is False:
+            return False
+        # if dash_lbtc is False:
+        #     return False
+
+        rates = await self.get_rates_all()
+        cumul_usd_agora = await self.open_trades_usd_parse_dash(
+            "agora", dash_agora, rates
+        )
+        # cumul_usd_lbtc = await self.open_trades_usd_parse_dash("lbtc", dash_lbtc,
+        # rates)
+        cumul_usd = cumul_usd_agora  # + cumul_usd_lbtc
+
+        cast_es = {
+            "trades_usd": cumul_usd,
+        }
+        await self.write_to_es("get_open_trades_usd", cast_es)
+        return cumul_usd
+
+    async def get_total_remaining(self):
+        """
+        Check how much profit we need to make in order to withdraw, taking into account
+         open trade value.
+        :return: profit remaining in USD
+        :rtype: float
+        """
+        total_usd = await self.get_total_usd()
+        total_trades_usd = await self.get_open_trades_usd()
+        if not total_usd:
+            return False
+        total_usd += total_trades_usd
+        withdraw_threshold = float(settings.Money.BaseUSD) + float(
+            settings.Money.WithdrawLimit
+        )
+        remaining = withdraw_threshold - total_usd
+
+        cast_es = {
+            "total_remaining_usd": remaining,
+        }
+        await self.write_to_es("get_total_remaining", cast_es)
+        return remaining
+
+    async def get_total_with_trades(self):
+        total_usd = await self.get_total_usd()
+        if not total_usd:
+            return False
+        total_trades_usd = await self.get_open_trades_usd()
+        total_with_trades = total_usd + total_trades_usd
+        cast_es = {
+            "total_with_trades": total_with_trades,
+        }
+        await self.write_to_es("get_total_with_trades", cast_es)
+        return total_with_trades
+
+
+money = Money()

core/lib/notify.py

@@ -1,4 +1,4 @@
-import requests
+import aiohttp
 
 from core.util import logs
 
@@ -8,7 +8,7 @@ log = logs.get_logger(__name__)
 
 
 # Actual function to send a message to a topic
-def raw_sendmsg(msg, title=None, priority=None, tags=None, url=None, topic=None):
+async def raw_sendmsg(msg, title=None, priority=None, tags=None, url=None, topic=None):
     if url is None:
         url = NTFY_URL
     headers = {"Title": "Pluto"}
@@ -18,15 +18,17 @@ def raw_sendmsg(msg, title=None, priority=None, tags=None, url=None, topic=None)
         headers["Priority"] = priority
     if tags:
         headers["Tags"] = tags
-    requests.post(
+    cast = {
-        f"{url}/{topic}",
+        "headers": headers,
-        data=msg,
+        "data": msg,
-        headers=headers,
+    }
-    )
+    async with aiohttp.ClientSession() as session:
+        async with session.post(f"{url}/{topic}", **cast) as response:
+            response = await response.content()
 
 
 # Sendmsg helper to send a message to a user's notification settings
-def sendmsg(user, *args, **kwargs):
+async def sendmsg(user, *args, **kwargs):
     notification_settings = user.get_notification_settings()
 
     if notification_settings.ntfy_topic is None:
@@ -35,4 +37,4 @@ def sendmsg(user, *args, **kwargs):
     else:
         topic = notification_settings.ntfy_topic
 
-    raw_sendmsg(*args, **kwargs, url=notification_settings.ntfy_url, topic=topic)
+    await raw_sendmsg(*args, **kwargs, url=notification_settings.ntfy_url, topic=topic)

core/migrations/0008_platform.py

@@ -0,0 +1,42 @@
+# Generated by Django 4.1.7 on 2023-03-09 20:50
+
+import uuid
+
+import django.db.models.deletion
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('core', '0007_alter_aggregator_account_info_and_more'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='Platform',
+            fields=[
+                ('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
+                ('name', models.CharField(max_length=255)),
+                ('service', models.CharField(choices=[('agora', 'Agora')], max_length=255)),
+                ('token', models.CharField(max_length=1024)),
+                ('password', models.CharField(max_length=1024)),
+                ('otp_token', models.CharField(blank=True, max_length=1024, null=True)),
+                ('username', models.CharField(max_length=255)),
+                ('send', models.BooleanField(default=True)),
+                ('cheat', models.BooleanField(default=False)),
+                ('dummy', models.BooleanField(default=False)),
+                ('cheat_interval_seconds', models.IntegerField(default=600)),
+                ('margin', models.FloatField(default=1.2)),
+                ('max_margin', models.FloatField(default=1.3)),
+                ('min_margin', models.FloatField(default=1.15)),
+                ('min_trade_size_usd', models.FloatField(default=10)),
+                ('max_trade_size_usd', models.FloatField(default=4000)),
+                ('accept_within_usd', models.FloatField(default=1)),
+                ('no_reference_amount_check_max_usd', models.FloatField(default=400)),
+                ('enabled', models.BooleanField(default=True)),
+                ('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL)),
+            ],
+        ),
+    ]

core/models.py

@@ -10,6 +10,8 @@ log = logs.get_logger(__name__)
 
 SERVICE_CHOICES = (("nordigen", "Nordigen"),)
 
+PLATFORM_SERVICE_CHOICES = (("agora", "Agora"),)
+
 
 class User(AbstractUser):
     id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
@@ -62,3 +64,68 @@ class Aggregator(models.Model):
     @property
     def client(self):
         pass
+
+    @classmethod
+    def get_for_platform(cls, platform):
+        return cls.objects.filter(user=platform.user, enabled=True)
+
+    @classmethod
+    def get_currencies_for_platform(cls, platform):
+        aggregators = Aggregator.get_for_platform(platform)
+        currencies = set()
+        for aggregator in aggregators:
+            for currency in aggregator.currencies:
+                currencies.add(currency)
+
+        return list(currencies)
+
+    @classmethod
+    def get_account_info_for_platform(cls, platform):
+        aggregators = Aggregator.get_for_platform(platform)
+        account_info = {}
+        for agg in aggregators:
+            for bank, accounts in agg.account_info.items():
+                if bank not in account_info:
+                    account_info[bank] = []
+                for account in accounts:
+                    account_info[bank].append(account)
+        return account_info
+
+
+class Platform(models.Model):
+    """
+    A connection to an arbitrage platform like AgoraDesk.
+    """
+
+    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
+    user = models.ForeignKey(User, on_delete=models.CASCADE)
+    name = models.CharField(max_length=255)
+    service = models.CharField(max_length=255, choices=PLATFORM_SERVICE_CHOICES)
+    token = models.CharField(max_length=1024)
+    password = models.CharField(max_length=1024)
+    otp_token = models.CharField(max_length=1024, null=True, blank=True)
+    username = models.CharField(max_length=255)
+
+    send = models.BooleanField(default=True)
+    cheat = models.BooleanField(default=False)
+    dummy = models.BooleanField(default=False)
+    cheat_interval_seconds = models.IntegerField(default=600)
+
+    margin = models.FloatField(default=1.20)
+    max_margin = models.FloatField(default=1.30)
+    min_margin = models.FloatField(default=1.15)
+    min_trade_size_usd = models.FloatField(default=10)
+    max_trade_size_usd = models.FloatField(default=4000)
+
+    accept_within_usd = models.FloatField(default=1)
+    no_reference_amount_check_max_usd = models.FloatField(default=400)
+
+    enabled = models.BooleanField(default=True)
+
+    @property
+    def currencies(self):
+        return Aggregator.get_currencies_for_platform(self)
+
+    @property
+    def account_info(self):
+        return Aggregator.get_account_info_for_platform(self)

core/templates/base.html

@@ -262,7 +262,7 @@
                 <a class="navbar-item" href="{% url 'aggregators' type='page' %}">
                   Bank Aggregators
                 </a>
-                <a class="navbar-item" href="#">
+                <a class="navbar-item" href="{% url 'platforms' type='page' %}">
                   Platform Connections
                 </a>
               </div>

core/templates/partials/platform-list.html

@@ -0,0 +1,81 @@
+{% load cache %}
+{% load cachalot cache %}
+{% get_last_invalidation 'core.Platform' as last %}
+{% include 'mixins/partials/notify.html' %}
+{# cache 600 objects_platforms request.user.id object_list type last #}
+<table
+  class="table is-fullwidth is-hoverable"
+  hx-target="#{{ context_object_name }}-table"
+  id="{{ context_object_name }}-table"
+  hx-swap="outerHTML"
+  hx-trigger="{{ context_object_name_singular }}Event from:body"
+  hx-get="{{ list_url }}">
+  <thead>
+    <th>id</th>
+    <th>user</th>
+    <th>name</th>
+    <th>service</th>
+    <th>enabled</th>
+    <th>actions</th>
+  </thead>
+  {% for item in object_list %}
+    <tr>
+      <td>
+        <a
+          class="has-text-grey"
+          onclick="window.prompt('Copy to clipboard: Ctrl+C, Enter', '{{ item.id }}/');">
+          <span class="icon" data-tooltip="Copy to clipboard">
+            <i class="fa-solid fa-copy" aria-hidden="true"></i>
+          </span>
+        </a>
+      </td>
+      <td>{{ item.user }}</td>
+      <td>{{ item.name }}</td>
+      <td>{{ item.get_service_display }}</td>
+      <td>
+        {% if item.enabled %}
+          <span class="icon">
+            <i class="fa-solid fa-check"></i>
+          </span>
+        {% else %}
+          <span class="icon">
+            <i class="fa-solid fa-xmark"></i>
+          </span>
+        {% endif %}
+      </td>
+      <td>
+        <div class="buttons">
+          <button
+            hx-headers='{"X-CSRFToken": "{{ csrf_token }}"}'
+            hx-get="{% url 'platform_update' type=type pk=item.id %}"
+            hx-trigger="click"
+            hx-target="#{{ type }}s-here"
+            hx-swap="innerHTML"
+            class="button">
+            <span class="icon-text">
+              <span class="icon">
+                <i class="fa-solid fa-pencil"></i>
+              </span>
+            </span>
+          </button>
+          <button
+            hx-headers='{"X-CSRFToken": "{{ csrf_token }}"}'
+            hx-delete="{% url 'platform_delete' type=type pk=item.id %}"
+            hx-trigger="click"
+            hx-target="#modals-here"
+            hx-swap="innerHTML"
+            hx-confirm="Are you sure you wish to delete {{ item.name }}?"
+            class="button">
+            <span class="icon-text">
+              <span class="icon">
+                <i class="fa-solid fa-xmark"></i>
+              </span>
+            </span>
+          </button>
+        </div>
+      </td>
+    </tr>
+  {% endfor %}
+
+</table>
+{# endcache #}

core/views/platforms.py

@@ -0,0 +1,50 @@
+from django.contrib.auth.mixins import LoginRequiredMixin
+from django.http import HttpResponse
+from django.shortcuts import render
+from django.urls import reverse
+from django.views import View
+from mixins.views import (  # ObjectRead,
+    ObjectCreate,
+    ObjectDelete,
+    ObjectList,
+    ObjectUpdate,
+)
+from two_factor.views.mixins import OTPRequiredMixin
+
+# from core.clients.platforms.agora import AgoraClient
+from core.forms import PlatformForm
+from core.models import Platform
+from core.util import logs
+
+# from core.views.helpers import synchronize_async_helper
+
+log = logs.get_logger(__name__)
+
+
+class PlatformList(LoginRequiredMixin, OTPRequiredMixin, ObjectList):
+    list_template = "partials/platform-list.html"
+    model = Platform
+    page_title = "List of platform connections"
+
+    list_url_name = "platforms"
+    list_url_args = ["type"]
+
+    submit_url_name = "platform_create"
+
+
+class PlatformCreate(LoginRequiredMixin, OTPRequiredMixin, ObjectCreate):
+    model = Platform
+    form_class = PlatformForm
+
+    submit_url_name = "platform_create"
+
+
+class PlatformUpdate(LoginRequiredMixin, OTPRequiredMixin, ObjectUpdate):
+    model = Platform
+    form_class = PlatformForm
+
+    submit_url_name = "platform_update"
+
+
+class PlatformDelete(LoginRequiredMixin, OTPRequiredMixin, ObjectDelete):
+    model = Platform

requirements.txt

@@ -25,10 +25,8 @@ redis
 hiredis
 django-cachalot
 PyOTP
-pycoingecko
+aiocoingecko
 requests
-arrow
-httpx
 forex_python
 pyOpenSSL
 Klein
@@ -36,3 +34,4 @@ ConfigObject
 aiohttp[speedups]
 elasticsearch[async]
 uvloop
+arrow