pluto/handler/ux/commands.py

620 lines
23 KiB
Python

# Other library imports
from json import dumps, loads
# Project imports
from settings import settings
class IRCCommands(object):
class trades(object):
name = "trades"
authed = True
helptext = "Get all open trades."
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
"""
Get details of open trades and post on IRC.
"""
# Send IRC - we don't want to automatically send messages on IRC, even though
# this variable seems counter-intuitive here, we are doing something with the result
# then calling msg() ourselves, and we don't want extra spam in the channel.
trades = agora.get_dashboard()
if not trades:
msg("No open trades.")
return
for trade_id in trades:
msg(trade_id)
class create(object):
name = "create"
authed = True
helptext = "Create an ad. Usage: create <XMR/BTC> <country> <currency> [<provider>]"
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
"""
Post an ad on AgoraDesk with the given country and currency code.
"""
if length == 4:
if spl[1] not in loads(settings.Agora.AssetList):
msg(f"Not a valid asset: {spl[1]}")
return
posted = agora.create_ad(spl[1], spl[2], spl[3], "REVOLUT")
if posted["success"]:
msg(f"{posted['response']['data']['message']}: {posted['response']['data']['ad_id']}")
else:
msg(dumps(posted["response"]))
elif length == 5:
if spl[1] not in loads(settings.Agora.AssetList):
msg(f"Not a valid asset: {spl[1]}")
return
if spl[4] not in loads(settings.Agora.ProviderList):
msg(f"Not a valid provider: {spl[4]}")
return
posted = agora.create_ad(spl[1], spl[2], spl[3], spl[4])
if posted["success"]:
msg(f"{posted['response']['data']['message']}: {posted['response']['data']['ad_id']}")
else:
msg(dumps(posted["response"]))
class messages(object):
name = "messages"
authed = True
helptext = "Get messages. Usage: messages [<reference>]"
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
"""
Get all messages for all open trades or a given trade.
"""
if length == 1:
messages = agora.get_recent_messages()
if messages is False:
msg("Error getting messages.")
return
if not messages:
msg("No messages.")
return
for reference in messages:
for message in messages[reference]:
msg(f"{reference}: {message[0]} {message[1]}")
msg("---")
elif length == 2:
tx = tx.ref_to_tx(spl[1])
if not tx:
msg(f"No such reference: {spl[1]}")
return
messages = agora.get_messages(spl[1], send_irc=False)
if not messages:
msg("No messages.")
for message in messages:
msg(f"{spl[1]}: {message}")
class dist(object):
name = "dist"
authed = True
helptext = "Distribute all our chosen currency and country ad pairs. Usage: dist [<XMR/BTC>]"
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
# Distribute out our ad to all countries in the config
if length == 2:
asset = spl[1]
if asset not in loads(settings.Agora.AssetList):
msg(f"Not a valid asset: {spl[1]}")
return
for x in agora.dist_countries(filter_asset=asset):
if x["success"]:
msg(f"{x['response']['data']['message']}: {x['response']['data']['ad_id']}")
else:
msg(dumps(x["response"]))
elif length == 1:
for x in agora.dist_countries():
if x["success"]:
msg(f"{x['response']['data']['message']}: {x['response']['data']['ad_id']}")
else:
msg(dumps(x["response"]))
class redist(object):
name = "redist"
authed = True
helptext = "Update all ads with details."
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
for x in agora.redist_countries():
if x[0]["success"]:
msg(f"{x[0]['response']['data']['message']}: {x[1]}")
else:
msg(dumps(x[0]["response"]))
class stripdupes(object):
name = "stripdupes"
authed = True
helptext = "Remove all duplicate adverts."
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
rtrn = agora.strip_duplicate_ads()
msg(dumps(rtrn))
class total(object):
name = "total"
authed = True
helptext = "Get total account balance from Sinks and Agora."
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
totals_all = tx.get_total()
totals = totals_all[0]
wallets = totals_all[1]
msg(f"Totals: SEK: {totals[0]} | USD: {totals[1]} | GBP: {totals[2]}")
msg(f"Wallets: XMR USD: {wallets[0]} | BTC USD: {wallets[1]}")
class ping(object):
name = "ping"
authed = False
helptext = "Pong!"
@staticmethod
def run(cmd, spl, length, authed, msg):
msg("Pong!")
class summon(object):
name = "summon"
authed = True
helptext = "Summon all operators."
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
ux.notify.sendmsg("You have been summoned!")
class message(object):
name = "msg"
authed = True
helptext = "Send a message on a trade. Usage: msg <reference> <message...>"
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
if length > 2:
full_msg = " ".join(spl[2:])
reference = tx.ref_to_tx(spl[1])
if not reference:
msg(f"No such reference: {spl[1]}")
return
rtrn = agora.agora.contact_message_post(reference, full_msg)
msg(f"Sent {full_msg} to {reference}: {rtrn}")
class refs(object):
name = "refs"
authed = True
helptext = "List all references"
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
msg(f"References: {', '.join(tx.get_refs())}")
class ref(object):
name = "ref"
authed = True
helptext = "Get more information about a reference. Usage: ref <reference>"
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
if length == 2:
ref_data = tx.get_ref(spl[1])
if not ref_data:
msg(f"No such reference: {spl[1]}")
return
msg(f"{spl[1]}: {dumps(ref_data)}")
class delete(object):
name = "del"
authed = True
helptext = "Delete a reference. Usage: del <reference>"
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
if length == 2:
ref_data = tx.get_ref(spl[1])
if not ref_data:
msg(f"No such reference: {spl[1]}")
return
tx.del_ref(spl[1])
msg(f"Deleted reference: {spl[1]}")
class release(object):
name = "release"
authed = True
helptext = "Release funds for a trade. Usage: release <reference>"
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
if length == 2:
tx = tx.ref_to_tx(spl[1])
if not tx:
msg(f"No such reference: {spl[1]}")
return
rtrn = agora.release_funds(tx)
message = rtrn["message"]
message_long = rtrn["response"]["data"]["message"]
msg(f"{message} - {message_long}")
class nuke(object):
name = "nuke"
authed = True
helptext = "Delete all our adverts."
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
rtrn = agora.nuke_ads()
msg(dumps(rtrn))
class wallet(object):
name = "wallet"
authed = True
helptext = "Get Agora wallet balances."
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
rtrn_xmr = agora.agora.wallet_balance_xmr()
if not rtrn_xmr["success"]:
msg("Error getting XMR wallet details.")
return
rtrn_btc = agora.agora.wallet_balance()
if not rtrn_btc["success"]:
msg("Error getting BTC wallet details.")
return
balance_xmr = rtrn_xmr["response"]["data"]["total"]["balance"]
balance_btc = rtrn_btc["response"]["data"]["total"]["balance"]
msg(f"XMR wallet balance: {balance_xmr}")
msg(f"BTC wallet balance: {balance_btc}")
class pubads(object):
name = "pubads"
authed = True
helptext = "View public adverts for Agora. Usage: pubads <XMR/BTC> <currency> [<provider,...>]"
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
if length == 3:
asset = spl[1]
if asset not in loads(settings.Agora.AssetList):
msg(f"Not a valid asset: {spl[1]}")
return
currency = spl[2]
rtrn = agora.get_all_public_ads(assets=[asset], currencies=[currency])
if not rtrn:
msg("No results.")
return
for ad in rtrn[currency]:
msg(f"({ad[0]}) {ad[1]} {ad[2]} {ad[3]} {ad[4]} {ad[5]} {ad[6]}")
elif length == 4:
asset = spl[1]
if asset not in loads(settings.Agora.AssetList):
msg(f"Not a valid asset: {spl[1]}")
return
providers = spl[3].split(",")
currency = spl[2]
rtrn = agora.get_all_public_ads(assets=[asset], currencies=[currency], providers=providers)
if not rtrn:
msg("No results.")
return
for ad in rtrn[currency]:
msg(f"({ad[0]}) {ad[1]} {ad[2]} {ad[3]} {ad[4]} {ad[5]} {ad[6]}")
class apubads(object):
name = "apubads"
authed = True
helptext = "View public adverts for LocalBitcoins. Usage: pubads <XMR/BTC> <currency> [<provider,...>]"
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
if length == 3:
asset = spl[1]
if asset not in loads(settings.LocalBitcoins.AssetList):
msg(f"Not a valid asset: {spl[1]}")
return
currency = spl[2]
rtrn = tx.lbtc.get_all_public_ads(assets=[asset], currencies=[currency])
if not rtrn:
msg("No results.")
return
for ad in rtrn[currency]:
msg(f"({ad[0]}) {ad[1]} {ad[2]} {ad[3]} {ad[4]} {ad[5]} {ad[6]}")
elif length == 4:
asset = spl[1]
if asset not in loads(settings.LocalBitcoins.AssetList):
msg(f"Not a valid asset: {spl[1]}")
return
providers = spl[3].split(",")
currency = spl[2]
rtrn = tx.lbtc.get_all_public_ads(assets=[asset], currencies=[currency], providers=providers)
if not rtrn:
msg("No results.")
return
for ad in rtrn[currency]:
msg(f"({ad[0]}) {ad[1]} {ad[2]} {ad[3]} {ad[4]} {ad[5]} {ad[6]}")
class cheat(object):
name = "cheat"
authed = True
helptext = "Cheat the markets by manipulating our prices to exploit people. Usage: cheat [<XMR/BTC>]"
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
if length == 1:
agora.run_cheat_in_thread()
msg("Running cheat in thread.")
elif length == 2:
asset = spl[1]
if asset not in loads(settings.Agora.AssetList):
msg(f"Not a valid asset: {spl[1]}")
return
agora.run_cheat_in_thread([asset])
msg(f"Running cheat in thread for {asset}.")
class cheatnext(object):
name = "cheatnext"
authed = True
helptext = "Run the next currency for cheat."
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
if length == 1:
asset = agora.run_cheat_in_thread()
msg(f"Running next asset for cheat in thread: {asset}")
class ads(object):
name = "ads"
authed = True
helptext = "Get all our ad regions for Agora"
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
ads = agora.enum_ads()
for ad in ads:
msg(f"({ad[0]}) {ad[1]} {ad[2]} {ad[3]} {ad[4]}")
class lads(object):
name = "lads"
authed = True
helptext = "Get all our ad regions for LocalBitcoins"
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
ads = tx.sources.lbtc.enum_ads()
for ad in ads:
msg(f"({ad[0]}) {ad[1]} {ad[2]} {ad[3]} {ad[4]}")
class xmr(object):
name = "xmr"
authed = True
helptext = "Get current XMR price."
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
xmr_prices = agora.cg.get_price(ids="monero", vs_currencies=["sek", "usd", "gbp"])
price_sek = xmr_prices["monero"]["sek"]
price_usd = xmr_prices["monero"]["usd"]
price_gbp = xmr_prices["monero"]["gbp"]
msg(f"SEK: {price_sek} | USD: {price_usd} | GBP: {price_gbp}")
class btc(object):
name = "btc"
authed = True
helptext = "Get current BTC price."
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
xmr_prices = agora.cg.get_price(ids="bitcoin", vs_currencies=["sek", "usd", "gbp"])
price_sek = xmr_prices["bitcoin"]["sek"]
price_usd = xmr_prices["bitcoin"]["usd"]
price_gbp = xmr_prices["bitcoin"]["gbp"]
msg(f"SEK: {price_sek} | USD: {price_usd} | GBP: {price_gbp}")
class withdraw(object):
name = "withdraw"
authed = True
helptext = "Take profit."
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
agora.withdraw_funds()
class remaining(object):
name = "r"
authed = True
helptext = "Show how much is left before we are able to withdraw funds."
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
remaining = tx.get_remaining()
msg(f"Remaining: {remaining}USD")
class total_remaining(object):
name = "tr"
authed = True
helptext = "Show how much is left before we are able to withdraw funds (including open trades)."
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
remaining = tx.get_total_remaining()
msg(f"Total remaining: {remaining}USD")
class tradetotal(object):
name = "tradetotal"
authed = True
helptext = "Get total value of all open trades in USD."
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
total = tx.get_open_trades_usd()
msg(f"Total trades: {total}USD")
class dollar(object):
name = "$"
authed = True
helptext = "Get total value of everything, including open trades."
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
total = tx.get_total_with_trades()
msg(f"${total}")
class profit(object):
name = "profit"
authed = True
helptext = "Get total profit."
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
total = tx.money.get_profit()
msg(f"Profit: {total}USD")
class tprofit(object):
name = "tprofit"
authed = True
helptext = "Get total profit with open trades."
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
total = tx.money.get_profit(True)
msg(f"Profit: {total}USD")
class signin(object):
name = "signin"
authed = True
helptext = "Generate a TrueLayer signin URL. Usage: signin <account>"
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
if length == 2:
account = spl[1]
auth_url = tx.truelayer.create_auth_url(account)
msg(f"Auth URL for {account}: {auth_url}")
class nsignin(object):
name = "nsignin"
authed = True
helptext = "Generate a Nordigen signin URL. Usage: nsignin <country> <bank>"
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
if length == 3:
country = spl[1]
bank_name = spl[2]
auth_url = tx.sinks.nordigen.create_auth_url(country, bank_name)
if not auth_url:
msg("Could not find bank.")
return
msg(f"Auth URL for {bank_name}: {auth_url}")
class accounts(object):
name = "accounts"
authed = True
helptext = "Get a list of acccounts from TrueLayer. Usage: accounts <account>"
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
if length == 2:
account = spl[1]
accounts = tx.sinks.truelayer.get_accounts(account)
for account in accounts["results"]:
msg(f"{account['account_id']} {account['display_name']} {account['currency']}")
class naccounts(object):
name = "naccounts"
authed = True
helptext = "Get a list of acccounts from Nordigen. Usage: naccounts"
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
if length == 1:
accounts = tx.sinks.nordigen.get_all_account_info()
for name, accounts in accounts.items():
for account in accounts:
msg(f"{name} {account['account_id']} {account['details']} {account['currency']}")
class transactions(object):
name = "transactions"
authed = True
helptext = "Get a list of transactions. Usage: transactions <account> <account_id>"
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
if length == 3:
account = spl[1]
account_id = spl[2]
transactions = tx.sinks.truelayer.get_transactions(account, account_id)
for transaction in transactions:
txid = transaction["transaction_id"]
ptxid = transaction["meta"]["provider_transaction_id"]
txtype = transaction["transaction_type"]
timestamp = transaction["timestamp"]
amount = transaction["amount"]
currency = transaction["currency"]
description = transaction["description"]
msg(f"{timestamp} {txid} {ptxid} {txtype} {amount}{currency} {description}")
class mapaccount(object):
name = "mapaccount"
authed = True
helptext = "Enable an account_id at a bank for use in TrueLayer. Usage: mapaccount <bank> <account_id>"
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
if length == 3:
bank = spl[1]
account_id = spl[2]
account_name = tx.sinks.truelayer.map_account(bank, account_id)
if not account_name:
msg(f"Failed to map the account")
return
msg(f"Mapped account ID {account_id} at bank {bank} to {account_name}")
class nmapaccount(object):
name = "nmapaccount"
authed = True
helptext = "Enable an account_id at a bank for use in Nordigen. Usage: nmapaccount <bank> <account_id>"
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
if length == 2:
account_id = spl[1]
account_name = tx.sinks.nordigen.map_account(account_id)
if not account_name:
msg(f"Failed to map the account")
return
msg(f"Mapped account ID {account_id} to {account_name}")
class unmapped(object):
name = "unmapped"
authed = True
helptext = "Get unmapped accounts for a bank. Usage: unmapped <bank>"
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
if length == 2:
bank = spl[1]
accounts_active = []
for bank, accounts in tx.sinks.truelayer.banks.items():
for account in accounts:
accounts_active.append(account)
accounts_all = tx.sinks.truelayer.get_accounts(bank)
accounts_unmapped = [x["account_id"] for x in accounts_all["results"] if x["account_id"] not in accounts_active]
msg(f"Unmapped accounts: {', '.join(accounts_unmapped)}")
class distdetails(object):
name = "distdetails"
authed = True
helptext = "Distribute account details among all ads."
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
currencies = tx.sinks.currencies
tx.markets.distribute_account_details()
msg(f"Distributing account details for currencies: {', '.join(currencies)}")