-
-
Notifications
You must be signed in to change notification settings - Fork 493
/
decoder.py
174 lines (151 loc) · 8.14 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
167
168
169
170
171
172
173
174
import json
import os
from typing import TYPE_CHECKING, Any
from rotkehlchen.accounting.structures.balance import Balance
from rotkehlchen.accounting.structures.types import HistoryEventSubType, HistoryEventType
from rotkehlchen.chain.ethereum.utils import asset_normalized_value
from rotkehlchen.chain.evm.contracts import EvmContract
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.types import ChecksumEvmAddress
from .constants import CPT_DXDAO_MESA
if TYPE_CHECKING:
from rotkehlchen.chain.ethereum.node_inquirer import EthereumInquirer
from rotkehlchen.chain.evm.decoding.base import BaseDecoderTools
from rotkehlchen.user_messages import MessagesAggregator
DEPOSIT = b'\xc1\x1c\xc3N\x93\xc6z\x938+\x99\xf2I\x8e\x997\x19\x87\x98\xf3\xc1\xc2\x88\x80\x08\xff\xc0\xee\xb8/h\xc4' # noqa: E501
ORDER_PLACEMENT = b'\xde\xcfo\xde\x82C\x98\x12\x99\xf7\xb7\xa7v\xf2\x9a\x9f\xc6z,\x98H\xe2]w\xc5\x0e\xb1\x1f\xa5\x8a~!' # noqa: E501
WITHDRAW_REQUEST = b',bE\xafPo\x0f\xc1\x08\x99\x18\xc0,\x1d\x01\xbd\xe9\xcc\x80v\t\xb34\xb3\xe7dMm\xfbZl^' # noqa: E501
WITHDRAW = b'\x9b\x1b\xfa\x7f\xa9\xeeB\n\x16\xe1$\xf7\x94\xc3Z\xc9\xf9\x04r\xac\xc9\x91@\xeb/dG\xc7\x14\xca\xd8\xeb' # noqa: E501
class DxdaomesaDecoder(DecoderInterface):
def __init__( # pylint: disable=super-init-not-called
self,
ethereum_inquirer: 'EthereumInquirer',
base_tools: 'BaseDecoderTools',
msg_aggregator: 'MessagesAggregator',
) -> None:
super().__init__(
evm_inquirer=ethereum_inquirer,
base_tools=base_tools,
msg_aggregator=msg_aggregator,
)
dir_path = os.path.dirname(os.path.realpath(__file__))
with open(os.path.join(dir_path, 'data', 'contracts.json'), encoding='utf8') as f:
contracts = json.loads(f.read())
self.contract = EvmContract(
address=contracts['DXDAOMESA']['address'],
abi=contracts['DXDAOMESA']['abi'],
deployed_block=contracts['DXDAOMESA']['deployed_block'],
)
def _decode_events(self, context: DecoderContext) -> DecodingOutput:
if context.tx_log.topics[0] == DEPOSIT:
return self._decode_deposit(context=context)
if context.tx_log.topics[0] == ORDER_PLACEMENT:
return self._decode_order_placement(context=context)
if context.tx_log.topics[0] == WITHDRAW_REQUEST:
return self._decode_withdraw_request(context=context)
if context.tx_log.topics[0] == WITHDRAW:
return self._decode_withdraw(context=context)
return DEFAULT_DECODING_OUTPUT
def _decode_deposit(self, context: DecoderContext) -> DecodingOutput:
topic_data, log_data = self.contract.decode_event(
tx_log=context.tx_log,
event_name='Deposit',
argument_names=('user', 'token', 'amount', 'batchId'),
)
deposited_asset = self.base.get_or_create_evm_asset(topic_data[1])
amount = asset_normalized_value(amount=log_data[0], asset=deposited_asset)
for event in context.decoded_events:
# Find the transfer event which should come before the deposit
if event.event_type == HistoryEventType.SPEND and event.asset == deposited_asset and event.balance.amount == amount and event.address == self.contract.address: # noqa: E501
event.event_type = HistoryEventType.DEPOSIT
event.event_subtype = HistoryEventSubType.DEPOSIT_ASSET
event.counterparty = CPT_DXDAO_MESA
event.notes = f'Deposit {amount} {deposited_asset.symbol} to DXDao mesa exchange'
break
return DEFAULT_DECODING_OUTPUT
def _decode_withdraw(self, context: DecoderContext) -> DecodingOutput:
topic_data, log_data = self.contract.decode_event(
tx_log=context.tx_log,
event_name='Withdraw',
argument_names=('user', 'token', 'amount'),
)
withdraw_asset = self.base.get_or_create_evm_asset(topic_data[1])
amount = asset_normalized_value(amount=log_data[0], asset=withdraw_asset)
for event in context.decoded_events:
# Find the transfer event which should come before the withdraw
if event.event_type == HistoryEventType.RECEIVE and event.asset == withdraw_asset and event.balance.amount == amount and event.address == self.contract.address: # noqa: E501
event.event_type = HistoryEventType.WITHDRAWAL
event.event_subtype = HistoryEventSubType.REMOVE_ASSET
event.counterparty = CPT_DXDAO_MESA
event.notes = f'Withdraw {amount} {withdraw_asset.symbol} from DXDao mesa exchange'
break
return DEFAULT_DECODING_OUTPUT
def _decode_withdraw_request(self, context: DecoderContext) -> DecodingOutput:
topic_data, log_data = self.contract.decode_event(
tx_log=context.tx_log,
event_name='WithdrawRequest',
argument_names=('user', 'token', 'amount', 'batchId'),
)
user = topic_data[0]
if not self.base.is_tracked(user):
return DEFAULT_DECODING_OUTPUT
token = self.base.get_or_create_evm_asset(topic_data[1])
amount = asset_normalized_value(amount=log_data[0], asset=token)
event = self.base.make_event_from_transaction(
transaction=context.transaction,
tx_log=context.tx_log,
event_type=HistoryEventType.INFORMATIONAL,
event_subtype=HistoryEventSubType.REMOVE_ASSET,
location_label=user,
asset=token,
balance=Balance(amount=amount),
notes=f'Request a withdrawal of {amount} {token.symbol} from DXDao Mesa',
counterparty=CPT_DXDAO_MESA,
address=context.transaction.to_address,
)
return DecodingOutput(event=event)
def _decode_order_placement(self, context: DecoderContext) -> DecodingOutput:
"""Some docs: https://docs.gnosis.io/protocol/docs/tutorial-limit-orders/"""
topic_data, log_data = self.contract.decode_event(
tx_log=context.tx_log,
event_name='OrderPlacement',
argument_names=('owner', 'index', 'buyToken', 'sellToken', 'validFrom', 'validUntil', 'priceNumerator', 'priceDenominator'), # noqa: E501
)
owner = topic_data[0]
if not self.base.is_tracked(owner):
return DEFAULT_DECODING_OUTPUT
result = self.evm_inquirer.multicall_specific(
contract=self.contract,
method_name='tokenIdToAddressMap',
arguments=[[topic_data[1]], [topic_data[2]]],
) # The resulting addresses are non checksumed but they can be found in the DB
buy_token = self.base.get_or_create_evm_asset(result[0][0])
sell_token = self.base.get_or_create_evm_asset(result[1][0])
buy_amount = asset_normalized_value(amount=log_data[3], asset=buy_token)
sell_amount = asset_normalized_value(amount=log_data[4], asset=sell_token)
event = self.base.make_event_from_transaction(
transaction=context.transaction,
tx_log=context.tx_log,
location_label=owner,
event_type=HistoryEventType.INFORMATIONAL,
event_subtype=HistoryEventSubType.PLACE_ORDER,
asset=sell_token,
balance=Balance(amount=sell_amount),
notes=f'Place an order in DXDao Mesa to sell {sell_amount} {sell_token.symbol} for {buy_amount} {buy_token.symbol}', # noqa: E501
counterparty=CPT_DXDAO_MESA,
address=context.transaction.to_address,
)
return DecodingOutput(event=event)
def addresses_to_decoders(self) -> dict[ChecksumEvmAddress, tuple[Any, ...]]:
return {
self.contract.address: (self._decode_events,),
}
@staticmethod
def counterparties() -> tuple[CounterpartyDetails, ...]:
return (CounterpartyDetails(identifier=CPT_DXDAO_MESA, label='dxdao', image='dxdao.svg'),)