Simple discord bot - posts to discord channel when mining pool wins block

announce_wins_discord_bot.py

import base64
import json
import logging
import time

import click
import requests
from discord_webhook import DiscordWebhook

GC_PROJECT_ID = "1019088744576"
logging.basicConfig(level=logging.INFO)


def post_to_discord(webhook_endpoint: str, message: str) -> None:
    """post message to supplied webhook endpoint"""

    webhook = DiscordWebhook(url=webhook_endpoint, content=message)
    response = webhook.execute()


def get_wallet_mining_history(wallet_address: str):
    wallet_history_endpoint = f"https://mining.nyc/blocks?miner={wallet_address}"
    return requests.get(wallet_history_endpoint).json()


def get_last_won_block_secret_id(prod: bool) -> str:
    last_won_block_secret_id = "nyc-block-win-notifier-last-won-block"
    if not prod:
        last_won_block_secret_id += "-test"

    return last_won_block_secret_id


def get_last_recorded_win_block(prod: bool) -> int:
    return get_secret(get_last_won_block_secret_id(prod))


def record_win_block(blockheight: int, prod: bool) -> None:

    if prod:
        secret_id = get_last_won_block_secret_id(prod)
        logging.info(f"recording blockheight {blockheight} at {secret_id}")

        from google.cloud import secretmanager
        client = secretmanager.SecretManagerServiceClient()
        parent = client.secret_path(GC_PROJECT_ID, secret_id)
        payload = str(blockheight).encode("UTF-8")

        # Add the secret version.
        response = client.add_secret_version(request={"parent": parent, "payload": {"data": payload}})

        # Print the new secret version name.
        logging.info("Added secret version: {}".format(response.name))
    else:
        logging.info(f"test mode, skipping recording, would've recorded blockheight {blockheight}")


def get_secret(secret_id: str) -> str:
    logging.info(f"fetching secret from secret manager: {secret_id}")

    from google.cloud import secretmanager
    client = secretmanager.SecretManagerServiceClient()
    secret_name = f"projects/{GC_PROJECT_ID}/secrets/{secret_id}/versions/latest"
    secret = client.access_secret_version(request={"name": secret_name}).payload.data.decode("UTF-8")

    return secret

def get_pool_total() -> float:
    return 1048369.049585 #TODO fetch dynamically

def check_wins(wallet_address: str, starting_blockheight: int, last_won_block: int = None, prod: bool = False):

    # fetch history
    retries = 5
    while retries > 0:

        history_json = get_wallet_mining_history(wallet_address=wallet_address)
        retries = retries - 1

        if len(history_json) == 0:
            logging.info(f"history endpont returned 0 length result, {retries} retries left")
            if retries == 0:
                raise "Max retries reached, exiting"
        else:
            break

    current_block_endpoint = 'https://mining.nyc/blocks/current_block'
    current_block = requests.get(current_block_endpoint).json()['currentBlock']


    # process history
    blocks_won = {}
    blocks_lost = {}
    stx_commited = 0

    for blockheight, block in list(history_json.items()):

        # remove blocks that the wallet did not mine
        if wallet_address not in block['miners'].keys():
            del history_json[blockheight]
            continue

        # remove blocks before starting block for a pool
        if int(blockheight) < starting_blockheight:
            del history_json[blockheight]
            continue

        # skip blocks after current block for win and loss counts
        if int(blockheight) > current_block:
            continue

        if "winner" in block:

            if block["winner"] == wallet_address:
                # wallet won the block
                blocks_won[blockheight] = block
            else:
                # wallet lost the block
                blocks_lost[blockheight] = block

            # in either case count how much the wallet committed
            stx_commited += block["miners"][wallet_address]


    newest_won_blockheight = max(blocks_won.keys())

    # is this a new win?
    last_won_block = get_last_recorded_win_block(prod=prod) if last_won_block is None else last_won_block
    if newest_won_blockheight == last_won_block:
        post_content = None

    else:

        winning_block = blocks_won[newest_won_blockheight]
        winning_bid = winning_block["miners"][wallet_address] / 1000000
        winning_block_total_bid = sum(winning_block["miners"].values()) / 1000000
        winning_probability = 100 * winning_bid / winning_block_total_bid

        nycc_won = 250000 * len(blocks_won) #TODO This should change to match issuance schedule https://docs.citycoins.co/citycoins-core-protocol/issuance-schedule
        cost_basis = stx_commited / 1000000 / nycc_won
        winnings_per_100_stx = nycc_won / get_pool_total() * 100 * 0.96
        
        # compose message to post to discord
        post_content = "\n".join(
            [
                "\n\n\n",
                f"Pool 1 wins another Block - `#{newest_won_blockheight}`",
                
                f":trophy: Winnings:",
                f"```",
                f"{len(blocks_won)} Total blocks won | {nycc_won} Total NYCC won | at {cost_basis:.4f} STX/NYCC",
                f"100 STX contribution to the pool wins {winnings_per_100_stx:.2f} after fees",
                f"```",

                f":ice_cube: Block Counts:",
                f"```",
                f"{len(blocks_won):6} Won | {len(blocks_lost):6} Lost | {len(history_json) - len(blocks_won) - len(blocks_lost):6} Pending = {len(history_json):6} Total Mined",
                f"```",
                
                f":hammer: Bidding:",
                f"```",
                f"{stx_commited/1000000:.1f} STX spent | {get_pool_total():.1f} STX remaining | {stx_commited/1000000/(len(blocks_won)+len(blocks_lost)):.1f} STX Average bid",
                f"```",
            ]
        )

    return post_content, newest_won_blockheight


# gcf start
def gcf_start(event, context):
    """Triggered from a message on a Cloud Pub/Sub topic.
    Args:
         event (dict): Event payload.
         context (google.cloud.functions.Context): Metadata for the event.
    """
    pubsub_message = base64.b64decode(event["data"]).decode("utf-8")
    prod = json.loads(pubsub_message)['prod']
    logging.info(f"Prod Mode: {prod}")

    wallet_address = get_secret(secret_id="nyc-block-win-notifier-wallet-address")
    starting_blockheight = int(get_secret("nyc-block-win-notifier-starting-block"))
    
    response, newest_won_blockheight = check_wins(wallet_address=wallet_address, starting_blockheight=starting_blockheight)

    if response:

        logging.info(response)

        test_discord_webhook = get_secret("nyc-block-win-notifier-test-discord-webhook")
        post_to_discord(webhook_endpoint=test_discord_webhook, message=response)

        if prod:
            test_discord_webhook = get_secret(
                "nyc-block-win-notifier-prod-discord-webhook"
            )
            post_to_discord(webhook_endpoint=prod_discord_webhook, message=response)

            record_win_block(blockheight=newest_won_blockheight, prod=prod)



if __name__ == '__main__':

    # integration test, fetches values from GCP secret manager, requires GCP credentials
    gcf_start(
        {
            "data": base64.b64encode('{"prod": false}'.encode("utf-8")),
        }, 
        context=None,
    )

    # local test
    # response, newest_won_blockheight = check_wins(wallet_address="----", starting_blockheight=38825, last_won_block=38841)
    # logging.info(response)