-
-
Notifications
You must be signed in to change notification settings - Fork 521
/
decoder.py
147 lines (128 loc) · 8.37 KB
/
decoder.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
from typing import Any
from rotkehlchen.accounting.structures.types import HistoryEventSubType, HistoryEventType
from rotkehlchen.chain.ethereum.modules.aave.common import asset_to_atoken
from rotkehlchen.chain.ethereum.utils import asset_normalized_value
from rotkehlchen.chain.evm.constants import ZERO_ADDRESS
from rotkehlchen.chain.evm.decoding.interfaces import DecoderInterface
from rotkehlchen.chain.evm.decoding.structures import (
DEFAULT_DECODING_OUTPUT,
DecoderContext,
DecodingOutput,
)
from rotkehlchen.chain.evm.decoding.types import CounterpartyDetails
from rotkehlchen.chain.evm.decoding.utils import maybe_reshuffle_events
from rotkehlchen.chain.evm.types import string_to_evm_address
from rotkehlchen.types import ChecksumEvmAddress
from rotkehlchen.utils.misc import hex_or_bytes_to_address, hex_or_bytes_to_int
from ..constants import AAVE_LABEL, CPT_AAVE_V1
DEPOSIT = b'\xc1,W\xb1\xc7:,:.\xa4a>\x94v\xab\xb3\xd8\xd1F\x85z\xabs)\xe2BC\xfbYq\x0c\x82'
REDEEM_UNDERLYING = b'\x9cN\xd5\x99\xcd\x85U\xb9\xc1\xe8\xcdvC$\r}q\xebv\xb7\x92\x94\x8cI\xfc\xb4\xd4\x11\xf7\xb6\xb3\xc6' # noqa: E501
LIQUIDATION_CALL = b'V\x86GW\xfd[\x1f\xc9\xf3\x8f_:\x98\x1c\xd8\xaeQ,\xe4\x1b\x90,\xf7?\xc5\x06\xee6\x9ck\xc27' # noqa: E501
class Aavev1Decoder(DecoderInterface):
def _decode_pool_event(self, context: DecoderContext) -> DecodingOutput:
if context.tx_log.topics[0] == DEPOSIT:
return self._decode_deposit_event(context=context)
if context.tx_log.topics[0] == REDEEM_UNDERLYING:
return self._decode_redeem_underlying_event(context=context)
if context.tx_log.topics[0] == LIQUIDATION_CALL:
return self._decode_liquidation(context=context)
return DEFAULT_DECODING_OUTPUT
def _decode_deposit_event(self, context: DecoderContext) -> DecodingOutput:
reserve_address = hex_or_bytes_to_address(context.tx_log.topics[1])
reserve_asset = self.base.get_or_create_evm_asset(reserve_address)
user_address = hex_or_bytes_to_address(context.tx_log.topics[2])
raw_amount = hex_or_bytes_to_int(context.tx_log.data[0:32])
amount = asset_normalized_value(raw_amount, reserve_asset)
atoken = asset_to_atoken(asset=reserve_asset, version=1)
if atoken is None:
return DEFAULT_DECODING_OUTPUT
deposit_event = receive_event = None
for event in context.decoded_events:
if event.event_type == HistoryEventType.SPEND and event.location_label == user_address and amount == event.balance.amount and reserve_asset == event.asset: # noqa: E501
# find the deposit transfer (can also be an ETH internal transfer)
event.event_type = HistoryEventType.DEPOSIT
event.event_subtype = HistoryEventSubType.DEPOSIT_ASSET
event.counterparty = CPT_AAVE_V1
event.notes = f'Deposit {amount} {reserve_asset.symbol} to aave-v1 from {event.location_label}' # noqa: E501
deposit_event = event
elif event.event_type == HistoryEventType.RECEIVE and event.location_label == user_address and amount == event.balance.amount and atoken == event.asset: # noqa: E501
# find the receive aToken transfer
event.event_subtype = HistoryEventSubType.RECEIVE_WRAPPED
event.counterparty = CPT_AAVE_V1
event.notes = f'Receive {amount} {atoken.symbol} from aave-v1 for {event.location_label}' # noqa: E501
receive_event = event
elif event.event_type == HistoryEventType.RECEIVE and event.location_label == user_address and event.address == ZERO_ADDRESS and event.asset == atoken: # noqa: E501
event.event_subtype = HistoryEventSubType.REWARD
event.counterparty = CPT_AAVE_V1
event.notes = f'Gain {event.balance.amount} {atoken.symbol} from aave-v1 as interest' # noqa: E501
maybe_reshuffle_events(
ordered_events=[deposit_event, receive_event],
events_list=context.decoded_events,
)
return DEFAULT_DECODING_OUTPUT
def _decode_redeem_underlying_event(self, context: DecoderContext) -> DecodingOutput:
reserve_address = hex_or_bytes_to_address(context.tx_log.topics[1])
reserve_asset = self.base.get_or_create_evm_asset(reserve_address)
user_address = hex_or_bytes_to_address(context.tx_log.topics[2])
raw_amount = hex_or_bytes_to_int(context.tx_log.data[0:32])
amount = asset_normalized_value(raw_amount, reserve_asset)
atoken = asset_to_atoken(asset=reserve_asset, version=1)
if atoken is None:
return DEFAULT_DECODING_OUTPUT
receive_event = return_event = interest_event = None
for event in context.decoded_events:
if event.event_type == HistoryEventType.RECEIVE and event.location_label == user_address and amount == event.balance.amount and reserve_asset == event.asset: # noqa: E501
event.event_type = HistoryEventType.WITHDRAWAL
event.event_subtype = HistoryEventSubType.REMOVE_ASSET
event.counterparty = CPT_AAVE_V1
event.notes = f'Withdraw {amount} {reserve_asset.symbol} from aave-v1'
receive_event = event
elif event.event_type == HistoryEventType.SPEND and event.location_label == user_address and amount == event.balance.amount and atoken == event.asset: # noqa: E501
# find the redeem aToken transfer
event.event_type = HistoryEventType.SPEND
event.event_subtype = HistoryEventSubType.RETURN_WRAPPED
event.counterparty = CPT_AAVE_V1
event.notes = f'Return {amount} {atoken.symbol} to aave-v1'
return_event = event
elif event.event_type == HistoryEventType.RECEIVE and event.location_label == user_address and event.address == ZERO_ADDRESS and event.asset == atoken: # noqa: E501
event.event_subtype = HistoryEventSubType.REWARD
event.counterparty = CPT_AAVE_V1
event.notes = f'Gain {event.balance.amount} {atoken.symbol} from aave-v1 as interest' # noqa: E501
interest_event = event
maybe_reshuffle_events(
ordered_events=[return_event, receive_event, interest_event],
events_list=context.decoded_events,
)
return DEFAULT_DECODING_OUTPUT
def _decode_liquidation(self, context: DecoderContext) -> DecodingOutput:
"""
Decode AAVE v1 liquidations. When a liquidation happens the user returns the debt token.
"""
if self.base.is_tracked(hex_or_bytes_to_address(context.tx_log.topics[3])) is False:
return DEFAULT_DECODING_OUTPUT
for event in context.decoded_events:
asset = event.asset.resolve_to_evm_token()
if event.event_type == HistoryEventType.SPEND and asset_normalized_value(
amount=hex_or_bytes_to_int(context.tx_log.data[32:64]), # debt amount
asset=asset,
) == event.balance.amount:
# we are transfering the debt token
event.event_subtype = HistoryEventSubType.PAYBACK_DEBT
event.notes = f'Payback {event.balance.amount} {asset.symbol} for an aave-v1 position' # noqa: E501
event.counterparty = CPT_AAVE_V1
event.address = context.tx_log.address
event.extra_data = {'is_liquidation': True} # adding this field to the decoded event to differenciate paybacks happening in liquidations. # noqa: E501
elif event.event_type == HistoryEventType.RECEIVE:
event.event_subtype = HistoryEventSubType.GENERATE_DEBT
event.counterparty = CPT_AAVE_V1
event.notes = f'Interest payment of {event.balance.amount} {asset.symbol} for aave-v1 position' # noqa: E501
event.address = context.tx_log.address
return DEFAULT_DECODING_OUTPUT
# -- DecoderInterface methods
def addresses_to_decoders(self) -> dict[ChecksumEvmAddress, tuple[Any, ...]]:
return {
string_to_evm_address('0x398eC7346DcD622eDc5ae82352F02bE94C62d119'): (self._decode_pool_event,), # AAVE_V1_LENDING_POOL # noqa: E501
}
@staticmethod
def counterparties() -> tuple[CounterpartyDetails, ...]:
return (CounterpartyDetails(identifier=CPT_AAVE_V1, label=AAVE_LABEL, image='aave.svg'),)