-
-
Notifications
You must be signed in to change notification settings - Fork 493
/
decoder.py
166 lines (144 loc) · 6.98 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
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
import logging
from typing import TYPE_CHECKING, Any, Callable
from rotkehlchen.accounting.structures.types import HistoryEventSubType, HistoryEventType
from rotkehlchen.assets.asset import EvmToken
from rotkehlchen.chain.ethereum.modules.balancer.constants import BALANCER_LABEL, CPT_BALANCER_V2
from rotkehlchen.chain.ethereum.modules.balancer.v2.constants import V2_SWAP, VAULT_ADDRESS
from rotkehlchen.chain.ethereum.utils import asset_normalized_value
from rotkehlchen.chain.evm.decoding.interfaces import DecoderInterface
from rotkehlchen.chain.evm.decoding.structures import (
DEFAULT_DECODING_OUTPUT,
FAILED_ENRICHMENT_OUTPUT,
ActionItem,
DecoderContext,
DecodingOutput,
EnricherContext,
TransferEnrichmentOutput,
)
from rotkehlchen.chain.evm.decoding.types import CounterpartyDetails
from rotkehlchen.constants.assets import A_ETH, A_WETH
from rotkehlchen.constants.resolver import ethaddress_to_identifier
from rotkehlchen.logging import RotkehlchenLogsAdapter
from rotkehlchen.types import ChecksumEvmAddress
from rotkehlchen.utils.misc import hex_or_bytes_to_address, hex_or_bytes_to_int
if TYPE_CHECKING:
from rotkehlchen.chain.evm.decoding.base import BaseDecoderTools
from rotkehlchen.chain.evm.node_inquirer import EvmNodeInquirer
from rotkehlchen.user_messages import MessagesAggregator
logger = logging.getLogger(__name__)
log = RotkehlchenLogsAdapter(logger)
class Balancerv2Decoder(DecoderInterface):
def __init__(
self,
evm_inquirer: 'EvmNodeInquirer',
base_tools: 'BaseDecoderTools',
msg_aggregator: 'MessagesAggregator',
) -> None:
super().__init__(
evm_inquirer=evm_inquirer,
base_tools=base_tools,
msg_aggregator=msg_aggregator,
)
self.eth = A_ETH.resolve_to_crypto_asset()
self.weth = A_WETH.resolve_to_evm_token()
def decode_swap_creation(self, context: DecoderContext) -> DecodingOutput:
"""
Decode swap in Balancer v2. At the beggining of the transaction a SWAP event is created
with the information of the tokens and amounts and later some transfers are executed.
We need to detect this swap event and then match the transferred amounts with the ones
in the swap event. A special case is the swap of ETH that is wrapped before being sent.
In this case the token is WETH but we have a tranfer of ETH from the user.
"""
if context.tx_log.topics[0] != V2_SWAP:
return DEFAULT_DECODING_OUTPUT
# The transfer event appears after the swap event, so we need to propagate information
from_token_address = hex_or_bytes_to_address(context.tx_log.topics[2])
to_token_address = hex_or_bytes_to_address(context.tx_log.topics[3])
amount_in = hex_or_bytes_to_int(context.tx_log.data[0:32])
amount_out = hex_or_bytes_to_int(context.tx_log.data[32:64])
# Create action item to propagate the information about the swap to the transfer enrichers
to_token = EvmToken(ethaddress_to_identifier(to_token_address))
to_amount = asset_normalized_value(
amount=amount_out,
asset=to_token,
)
action_item = ActionItem(
action='skip & keep',
from_event_type=HistoryEventType.RECEIVE,
from_event_subtype=HistoryEventSubType.NONE,
asset=to_token,
amount=to_amount,
to_event_type=None,
to_event_subtype=None,
to_counterparty=CPT_BALANCER_V2,
to_notes=None,
extra_data={
'from_token': from_token_address,
'amount_in': amount_in,
},
)
# When ETH is swapped it is wrapped to WETH and the ETH transfer happens before the SWAP
# event. We need to detect it if we haven't done it yet.
if len(context.action_items) == 0 and from_token_address == self.weth.evm_address:
# when swapping eth the transfer event appears before the V2_SWAP event so we need
# to check if the asset swapped was ETH or not.
amount_of_eth = asset_normalized_value(
amount=amount_in,
asset=self.eth,
)
for event in context.decoded_events:
if (
event.asset == A_ETH and event.balance.amount == amount_of_eth and
event.event_type == HistoryEventType.SPEND and
event.event_subtype == HistoryEventSubType.NONE
):
event.event_type = HistoryEventType.TRADE
event.event_subtype = HistoryEventSubType.SPEND
event.notes = f'Swap {event.balance.amount} {self.eth.symbol} in Balancer v2'
event.counterparty = CPT_BALANCER_V2
return DecodingOutput(action_items=[action_item])
def _maybe_enrich_balancer_v2_transfers(
self,
context: EnricherContext,
) -> TransferEnrichmentOutput:
"""
Enrich tranfer transactions to account for swaps in balancer v2 protocol.
May raise:
- UnknownAsset
- WrongAssetType
"""
if context.action_items is None or len(context.action_items) == 0 or context.transaction.to_address != VAULT_ADDRESS: # noqa: E501
return FAILED_ENRICHMENT_OUTPUT
if context.action_items[-1].extra_data is None:
return FAILED_ENRICHMENT_OUTPUT
asset = context.event.asset.resolve_to_evm_token()
if (
isinstance(context.action_items[-1].asset, EvmToken) is False or
context.action_items[-1].asset.evm_address != context.tx_log.address or # type: ignore[attr-defined] # mypy fails to understand that due the previous statmenet in the or this check won't be evaluated if the asset isn't a token
context.action_items[-1].amount != context.event.balance.amount
):
return FAILED_ENRICHMENT_OUTPUT
context.event.counterparty = CPT_BALANCER_V2
context.event.event_type = HistoryEventType.TRADE
if asset == context.event.asset:
context.event.event_subtype = HistoryEventSubType.RECEIVE
context.event.notes = f'Receive {context.event.balance.amount} {asset.symbol} from Balancer v2' # noqa: E501
else:
context.event.event_subtype = HistoryEventSubType.SPEND
return TransferEnrichmentOutput(matched_counterparty=CPT_BALANCER_V2)
# -- DecoderInterface methods
def addresses_to_decoders(self) -> dict[ChecksumEvmAddress, tuple[Any, ...]]:
return {
VAULT_ADDRESS: (self.decode_swap_creation,),
}
def enricher_rules(self) -> list[Callable]:
return [
self._maybe_enrich_balancer_v2_transfers,
]
@staticmethod
def counterparties() -> tuple[CounterpartyDetails, ...]:
return (CounterpartyDetails(
identifier=CPT_BALANCER_V2,
label=BALANCER_LABEL,
image='balancer.svg',
),)