-
-
Notifications
You must be signed in to change notification settings - Fork 494
/
utils.py
112 lines (94 loc) · 4.25 KB
/
utils.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
from collections.abc import Sequence
from typing import TYPE_CHECKING, Optional
from rotkehlchen.assets.asset import AssetWithSymbol
from rotkehlchen.chain.evm.decoding.types import CounterpartyDetails
from rotkehlchen.fval import FVal
from rotkehlchen.history.events.structures.types import HistoryEventSubType, HistoryEventType
from rotkehlchen.types import ChainID, ChecksumEvmAddress
if TYPE_CHECKING:
from rotkehlchen.chain.evm.structures import EvmTxReceiptLog
from rotkehlchen.history.events.structures.evm_event import EvmEvent
def maybe_reshuffle_events(
ordered_events: Sequence[Optional['EvmEvent']],
events_list: list['EvmEvent'],
) -> None:
"""Takes a list of events to order and makes sure that the sequence index of each
of them is in ascending order and that the events are consecutive in the
decoded events list.
This is for two reasons.
1. So that it appears uniformly in the UI
2. So that during accounting we know which type of event comes first in a swap-like event.
For example for swaps we expect two consecutive events with the first
being the out event and the second the in event,
The events are optional since it's also possible they may not be found.
"""
actual_events = [x for x in ordered_events if x is not None]
if len(actual_events) <= 1:
return # nothing to do
all_other_events = []
max_seq_index = -1
for event in events_list:
if event not in actual_events:
all_other_events.append(event)
if event.sequence_index > max_seq_index:
max_seq_index = event.sequence_index
for idx, event in enumerate(actual_events):
event.sequence_index = max_seq_index + idx + 1
events_list = all_other_events + actual_events
def bridge_prepare_data(
tx_log: 'EvmTxReceiptLog',
deposit_topics: Sequence[bytes],
source_chain: ChainID,
target_chain: ChainID,
from_address: ChecksumEvmAddress,
to_address: ChecksumEvmAddress,
) -> tuple[HistoryEventType, HistoryEventType, ChainID, ChainID, ChecksumEvmAddress]:
"""Method to prepare the bridge variables
`source_chain` is the chain where the current transaction is happening
`target_chain` is the chain on the other side of the bridge
When coming here the caller has to make sure that:
- tx_log topics is either in deposit_topics or else is a withdrawal
"""
# Determine whether it is a deposit or a withdrawal
if tx_log.topics[0] in deposit_topics:
expected_event_type = HistoryEventType.SPEND
expected_location_label = from_address
new_event_type = HistoryEventType.DEPOSIT
from_chain, to_chain = source_chain, target_chain
else: # withdrawal
expected_event_type = HistoryEventType.RECEIVE
expected_location_label = to_address
new_event_type = HistoryEventType.WITHDRAWAL
from_chain, to_chain = target_chain, source_chain
return expected_event_type, new_event_type, from_chain, to_chain, expected_location_label
def bridge_match_transfer(
event: 'EvmEvent',
from_address: ChecksumEvmAddress,
to_address: ChecksumEvmAddress,
from_chain: ChainID,
to_chain: ChainID,
amount: FVal,
asset: AssetWithSymbol,
expected_event_type: HistoryEventType,
new_event_type: HistoryEventType,
counterparty: CounterpartyDetails,
) -> None:
"""Action to take when matching a bridge transfer event"""
from_label, to_label = f' address {from_address}', f' address {to_address}'
if expected_event_type == HistoryEventType.SPEND:
if event.location_label == from_address:
from_label = ''
if to_address == from_address:
to_label = ''
elif expected_event_type == HistoryEventType.RECEIVE:
if event.location_label == to_address:
to_label = ''
if to_address == from_address:
from_label = ''
event.event_type = new_event_type
event.event_subtype = HistoryEventSubType.BRIDGE
event.counterparty = counterparty.identifier
event.notes = (
f'Bridge {amount} {asset.symbol} from {from_chain.label()}{from_label} to '
f'{to_chain.label()}{to_label} via {counterparty.label} bridge'
)