rollbot4/commands/commands/rollcoin.py

from dataclasses import dataclass, field
from typing import Optional
import random

from rollbot import as_command, initialize_data, RollbotFailure
from rollbot.injection import Data, Sender, Config, Const, Arg, Reply, OriginAdmin

# View
#   !wallet - shows your number of rollcoins, NFTs (Non-Functional Tamagotchis), and market balance
#   !blockchain - shows the contents of all wallets and the rollcoins in the treasury
# Generate
#   !mine - provides a sudoku that can be solved for rollcoins, which also adds to the treasury
#   !appraise - rollbot will purchase a post, based indirectly on number of likes received
# Spend
#   !tip - transfer rollcoins from one person to another
#   !gacha - insert a rollcoin (which enters the treasury), receive a random NFT
#   !donate - split an amount and donate it to everyone
# Market
#   !bet - put some amount of rollcoins into the market, some portion of which will be put in the treasury, evolves market
#   !cash - take some amount of rollcoins from the market, up to investment + the number of coins in the treasury, evolves market
# Admin
#   !deflate - deflate all coins by a power of 10
#   !brrr - print some number of coins into the treasury
#   !mine !clear - clear the current mining puzzle
#   !market !force - force the market to transition to a given state


@initialize_data
@dataclass
class RollcoinState:
    treasury: float
    mining_puzzle: Optional[list[list[int]]]
    market_state: int


@initialize_data
@dataclass
class RollcoinWallet:
    balance: float = 10
    holdings: float = 0
    cost_basis: float = 0
    nfts: list[str] = field(default_factory=list)

    def __str__(self):
        s = f"Wallet: {self.balance}\n"
        if self.holdings > 0 or self.cost_basis > 0:
            s += f"Investments: {self.holdings}\n\t(cost basis {self.cost_basis})\n"
        if len(self.nfts) > 0:
            s += f"NFTs:\n"
        for nft in self.nfts:
            s += f"\t{nft}\n"
        return s.strip()


SPECIAL_AMOUNTS = {
    "all": lambda limit: limit,
    "half": lambda limit: limit / 2,
    "frac": lambda limit: limit - int(limit),
}


def convert_amount(amount):
    if (conv := SPECIAL_AMOUNTS.get(amount.lower(), None)) is not None:
        return conv
    return float(amount)


GLOBAL_STATE_KEY = "ROLLCOIN_GLOBAL_STATE"

# injection values
State = Data(RollcoinState).For(Const(GLOBAL_STATE_KEY), treasury=100, mining_puzzle=None, market_state=0)
SenderWallet = Data(RollcoinWallet).For(Sender)
WalletLookup = Config("rollcoin.wallet_names")
MarketTransitions = Config("rollcoin.market.transitions")
MarketMultipliers = Config("rollcoin.market.multipliers")
MarketMessages = Config("rollcoin.market.messages")


@as_command
def wallet(sender_wallet: SenderWallet, reply: Reply):
    return f"You currently own...\n{sender_wallet}", reply


@as_command
async def blockchain(wallets: Data(RollcoinWallet), wallet_lookup: WalletLookup, state: State):
    response = f"Blockchain:\n\tTreasury: {state.treasury}\n\n"
    names = {v: k.title() for k, v in wallet_lookup.items()}
    async for (sender_id, wallet) in wallets.all():
        response += names.get(sender_id, f"Unnamed wallet {sender_id}") + ":\n"
        response += "\n".join("\t" + s for s in str(wallet).split("\n")) + "\n\n"
    return response.strip()


@as_command
async def tip(
    wallets: Data(RollcoinWallet),
    sender_id: Sender,
    sender_wallet: SenderWallet,
    wallet_lookup: WalletLookup,
    target_name: Arg(0, missing_msg="You must tell me who to tip!"),
    amount: Arg(1, convert=convert_amount, missing_msg="You must provide an amount to tip!", fail_msg="Could not parse {} as value"),
):
    if not isinstance(amount, float):
        # handle special converters
        amount = amount(sender_wallet.balance)

    if sender_wallet.balance == 0:
        return "Sorry! You don't have any rollcoins right now - try mining!"

    if amount > sender_wallet.balance:
        return f"Sorry! You only have {sender_wallet.balance} RollCoins available - try mining for more!"

    if amount <= 0:
        RollbotFailure.INVALID_ARGUMENTS.raise_exc(f"Amount must be positive, not {amount}")

    if (target_id := wallet_lookup.get(target_name.lower(), None)) is None:
        RollbotFailure.INVALID_ARGUMENTS.raise_exc(f"Cannot find wallet belonging to {target_name}")

    target_wallet = await wallets.load_or(target_id)
    sender_wallet.balance -= amount
    target_wallet.balance += amount
    await wallets.save(target_id, target_wallet)
    await wallets.save(sender_id, sender_wallet)

    return f"Done! {target_name} now has {target_wallet.balance} RollCoins!"


async def evolve_market(state, transitions, multipliers, wallet_store, state_store):
    state.market_state = random.choices(range(len(transitions)), weights=transitions[state.market_state], k=1)[0]
    await state_store.save(GLOBAL_STATE_KEY, state)
    multiplier = multipliers[state.market_state]
    async for (wallet_id, wallet) in wallet_store.all():
        wallet.holdings *= multiplier
        await wallet_store.save(wallet_id, wallet)


@as_command
async def bet(
    sender_id: Sender,
    sender_wallet: SenderWallet,
    wallet_store: Data(RollcoinWallet),
    state: State,
    state_store: Data(RollcoinState),
    transitions: MarketTransitions,
    multipliers: MarketMultipliers,
    messages: MarketMessages,
    amount: Arg(0, convert=convert_amount, missing_msg="You must provide an amount to bet!", fail_msg="Could not parse {} as value"),
    reply: Reply,
):
    if not isinstance(amount, float):
        # handle special converters
        amount = amount(sender_wallet.balance)

    if sender_wallet.balance == 0:
        yield "Sorry! You don't have any rollcoins right now - try mining!"
        return

    if amount > sender_wallet.balance:
        yield f"Sorry! You only have {sender_wallet.balance} RollCoins available - try mining for more!"
        return

    if amount <= 0:
        RollbotFailure.INVALID_ARGUMENTS.raise_exc(f"Amount must be positive, not {amount}")

    sender_wallet.balance -= amount
    sender_wallet.holdings += amount
    sender_wallet.cost_basis += amount
    await wallet_store.save(sender_id, sender_wallet)

    # coins enter treasury, allow evolve_market to save state
    state.treasury += amount
    await evolve_market(state, transitions, multipliers, wallet_store, state_store)

    yield f"Trade complete!\n{await wallet_store.load(sender_id)}", reply
    yield f"Market status: {messages[state.market_state]}"


@as_command
async def cash(
    sender_id: Sender,
    sender_wallet: SenderWallet,
    wallet_store: Data(RollcoinWallet),
    state: State,
    state_store: Data(RollcoinState),
    transitions: MarketTransitions,
    multipliers: MarketMultipliers,
    messages: MarketMessages,
    amount: Arg(0, convert=convert_amount, missing_msg="You must provide an amount to bet!", fail_msg="Could not parse {} as value"),
    reply: Reply,
):
    if not isinstance(amount, float):
        # handle special converters
        amount = amount(sender_wallet.holdings)

    if sender_wallet.holdings == 0:
        yield "Sorry! You don't have any rollcoins invested right now - try betting!"
        return

    if amount > sender_wallet.holdings:
        yield f"Sorry! You only have {sender_wallet.holdings} RollCoins in the market - try betting more first!"
        return

    if amount <= 0:
        RollbotFailure.INVALID_ARGUMENTS.raise_exc(f"Amount must be positive, not {amount}")

    if state.treasury == 0:
        yield f"Sorry! The treasury is actually empty right now so uh... no one can sell..."
        return

    if amount <= state.treasury:
        response = "Successfully sold!"
        actual_amount = amount
    else:
        response = f"Funny story, I'm actually out of coins! I gave you what I could, draining the treasury, which was {state.treasury}."
        actual_amount = state.treasury

    sender_wallet.balance += actual_amount
    sender_wallet.holdings -= actual_amount
    sender_wallet.cost_basis = max(sender_wallet.cost_basis - actual_amount, 0)
    await wallet_store.save(sender_id, sender_wallet)
    # coins exit treasury, allow evolve_market to save state
    state.treasury -= actual_amount
    await evolve_market(state, transitions, multipliers, wallet_store, state_store)

    yield f"{response}\n{await wallet_store.load(sender_id)}", reply
    yield f"Market status: {messages[state.market_state]}"


# ADMIN COMMANDS

@as_command
async def deflate(
    origin_admin: OriginAdmin,
    power: Arg(0, convert=int, missing_msg="Must provide power to deflate by", fail_msg="Power to deflate by must be an integer"),
    wallet_store: Data(RollcoinWallet),
    state_store: Data(RollcoinState),
):
    if not origin_admin:
        RollbotFailure.PERMISSIONS.raise_exc("Only admins can deflate the currency")

    factor = 10 ** power

    state = await state_store.load(GLOBAL_STATE_KEY)
    state.treasury /= factor
    await state_store.save(GLOBAL_STATE_KEY, state)
    
    async for (wallet_id, wallet) in wallet_store.all():
        wallet.balance /= factor
        wallet.holdings /= factor
        wallet.cost_basis /= factor
        await wallet_store.save(wallet_id, wallet)
    
    return f"Economy deflated by a factor of 10^{power}"