"""Deposit and withdraw detection and management."""
import logging
import dataclasses
import datetime
from decimal import Decimal
from typing import Dict, List
from dataclasses_json import dataclass_json
from eth_typing import HexAddress
from web3 import Web3
from eth_defi.balances import DecimalisedHolding, \
fetch_erc20_balances_by_token_list, convert_balances_to_decimal
from tradeexecutor.state.reserve import ReservePosition
from tradeexecutor.state.identifier import AssetIdentifier
logger = logging.getLogger(__name__)
[docs]@dataclass_json
@dataclasses.dataclass
class ReserveUpdateEvent:
"""A legacy reserve update event.
Maintained for old code compatibility.
See :py:mod:`tradeeexecutor.state.sync` for the current approach.
TODO: This should be removed as is partially part of old treasury sync code.
"""
asset: AssetIdentifier
#: Transfer timestamp (if known)
mined_at: datetime.datetime
#: Strategy cycle timestamp
updated_at: datetime.datetime
past_balance: Decimal
new_balance: Decimal
@property
def change(self) -> Decimal:
return self.new_balance - self.past_balance
[docs]def update_wallet_balances(web3: Web3, address: HexAddress, tokens: List[HexAddress]) -> Dict[HexAddress, DecimalisedHolding]:
"""Get raw balances of ERC-20 tokens."""
balances = fetch_erc20_balances_by_token_list(web3, address, tokens)
return convert_balances_to_decimal(web3, balances)
[docs]def sync_reserves(
web3: Web3,
clock: datetime.datetime,
wallet_address: HexAddress,
current_reserves: List[ReservePosition],
supported_reserve_currencies: List[AssetIdentifier]) -> List[ReserveUpdateEvent]:
"""Check the address for any incoming stablecoin transfers to see how much cash we have."""
our_chain_id = web3.eth.chain_id
# Get raw ERC-20 holdings of the address
balances = update_wallet_balances(web3, wallet_address, [web3.to_checksum_address(a.address) for a in supported_reserve_currencies])
reserves_per_token = {r.asset.address: r for r in current_reserves}
events: ReserveUpdateEvent = []
for currency in supported_reserve_currencies:
address = currency.address
# 1337 is Ganache
if our_chain_id != 1337:
assert currency.chain_id == our_chain_id, f"Asset expects chain_id {currency.chain_id}, currently connected to {our_chain_id}"
if currency.address in reserves_per_token:
# We have an existing record of having this reserve
current_value = reserves_per_token[address].quantity
else:
current_value = Decimal(0)
decimal_holding = balances.get(Web3.to_checksum_address(address))
# We get decimals = None if Ganache is acting
assert decimal_holding.decimals, f"Token did not have decimals: token:{currency} holding:{decimal_holding}"
if (decimal_holding is not None) and (decimal_holding.value != current_value):
evt = ReserveUpdateEvent(
asset=currency,
past_balance=current_value,
new_balance=decimal_holding.value,
updated_at=clock,
mined_at=clock, # TODO: We do not have logic to get actual block_mined_at of Transfer() here
)
events.append(evt)
logger.info("Reserve currency update detected. Asset: %s, past: %s, new: %s", evt.asset, evt.past_balance, evt.new_balance)
return events