pluto/handler/ux/notify.py

# Other library imports
import requests

# Project imports
from settings import settings
import util


class Notify(util.Base):
    """
    Class to handle more robust notifications.
    """

    def sendmsg(self, msg, title=None, priority=None, tags=None):
        headers = {"Title": "Bot"}
        if title:
            headers["Title"] = title
        if priority:
            headers["Priority"] = priority
        if tags:
            headers["Tags"] = tags
        requests.post(
            f"{settings.Notify.Host}/{settings.Notify.Topic}",
            data=msg,
            headers=headers,
        )

    def notify_new_trade(self, amount, currency):
        amount_usd = self.money.to_usd(amount, currency)
        self.sendmsg(f"Total: {amount_usd}", title="New trade", tags="trades", priority="2")

    def notify_complete_trade(self, amount, currency):
        amount_usd = self.money.to_usd(amount, currency)
        self.sendmsg(f"Total: {amount_usd}", title="Trade complete", tags="trades,profit", priority="3")

    def notify_withdrawal(self, amount_usd):
        self.sendmsg(f"Total: {amount_usd}", title="Withdrawal", tags="profit", priority="4")

    def notify_need_topup(self, amount_usd_xmr, amount_usd_btc):
        self.sendmsg(f"XMR: {amount_usd_xmr} | BTC: {amount_usd_btc}", title="Topup needed", tags="admin", priority="5")

    def notify_tx_lookup_failed(self, currency, amount, reference, code, trade_id=None):
        self.sendmsg(f"Unknown TX [{code}]: {amount}{currency} ({reference}) for {trade_id}", title=code, tags="tx", priority="5")

    def notify_release_unsuccessful(self, trade_id):
        self.sendmsg(f"Release unsuccessful for {trade_id}", title="Unsuccessful release", tags="tx", priority="5")