-
-
Notifications
You must be signed in to change notification settings - Fork 494
/
event.py
85 lines (66 loc) · 2.61 KB
/
event.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
from abc import ABCMeta, abstractmethod
from enum import auto
from typing import TYPE_CHECKING, Any
from rotkehlchen.assets.asset import Asset
from rotkehlchen.types import Timestamp
from rotkehlchen.utils.mixins.enums import SerializableEnumNameMixin
if TYPE_CHECKING:
from more_itertools import peekable
from rotkehlchen.accounting.pot import AccountingPot
from rotkehlchen.accounting.structures.types import ActionType
class AccountingEventType(SerializableEnumNameMixin):
TRADE = auto()
FEE = auto()
ASSET_MOVEMENT = auto()
MARGIN_POSITION = auto()
LOAN = auto()
PREFORK_ACQUISITION = auto()
STAKING = auto()
HISTORY_EVENT = auto()
TRANSACTION_EVENT = auto()
class AccountingEventMixin(metaclass=ABCMeta):
"""Interface to be followed by all data structures that go in accounting"""
@abstractmethod
def get_timestamp(self) -> Timestamp:
"""Get the event's timestamp"""
@staticmethod
@abstractmethod
def get_accounting_event_type() -> AccountingEventType:
"""
Returns the event type for accounting
"""
@abstractmethod
def get_identifier(self) -> str:
"""Get a unique identifier from an accounting event"""
@abstractmethod
def should_ignore(self, ignored_ids_mapping: dict['ActionType', set[str]]) -> bool:
"""Returns whether this event should be ignored due to user settings"""
@abstractmethod
def get_assets(self) -> list[Asset]:
"""Gets the assets involved in the event.
May raise:
- UnknownAsset, UnsupportedAsset due to the trade pair's assets
- UnprocessableTradePair: If a trade's pair can't be processed
"""
@abstractmethod
def process(
self,
accounting: 'AccountingPot',
events_iterator: "peekable['AccountingEventMixin']",
) -> int:
"""Processes the event for accounting and adds to the respective pot's processed events.
Can also accept the iterator of the processed events of the accounting pot if there is
a need to pull multiple events from the iterator (e.g. for defi swaps)
Returns the number of events consumed.
"""
@abstractmethod
def serialize(self) -> dict[str, Any]:
"""Serializes the event"""
def serialize_for_debug_import(self) -> dict[str, Any]:
data = self.serialize()
data['accounting_event_type'] = self.get_accounting_event_type().serialize()
return data
@classmethod
@abstractmethod
def deserialize(cls, data: dict[str, Any]) -> 'AccountingEventMixin':
"""Deserializes the event"""