9394559170

Committed 26 Apr 2024 06:36AM UTC coverage: 94.656% (-0.02%) from 94.674%

Build # 9394559170

Build Type

push

github

Committed by

xmatthias

Commit Message

Loader should be passed as kwarg for clarity

Run Details

20280 of 21425 relevant lines covered (94.66%)

0.95 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

86.32

/freqtrade/persistence/custom_data.py

import json
import logging
from datetime import datetime
from typing import Any, ClassVar, List, Optional, Sequence

from sqlalchemy import DateTime, ForeignKey, Integer, String, Text, UniqueConstraint, select
from sqlalchemy.orm import Mapped, mapped_column, relationship

from freqtrade.constants import DATETIME_PRINT_FORMAT
from freqtrade.persistence.base import ModelBase, SessionType
from freqtrade.util import dt_now


logger = logging.getLogger(__name__)


class _CustomData(ModelBase):
    """
    CustomData database model
    Keeps records of metadata as key/value store
    for trades or global persistent values
    One to many relationship with Trades:
      - One trade can have many metadata entries
      - One metadata entry can only be associated with one Trade
    """
    __tablename__ = 'trade_custom_data'
    __allow_unmapped__ = True
    session: ClassVar[SessionType]

    # Uniqueness should be ensured over pair, order_id
    # its likely that order_id is unique per Pair on some exchanges.
    __table_args__ = (UniqueConstraint('ft_trade_id', 'cd_key', name="_trade_id_cd_key"),)

    id = mapped_column(Integer, primary_key=True)
    ft_trade_id = mapped_column(Integer, ForeignKey('trades.id'), index=True)

    trade = relationship("Trade", back_populates="custom_data")

    cd_key: Mapped[str] = mapped_column(String(255), nullable=False)
    cd_type: Mapped[str] = mapped_column(String(25), nullable=False)
    cd_value: Mapped[str] = mapped_column(Text, nullable=False)
    created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, default=dt_now)
    updated_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True)

    # Empty container value - not persisted, but filled with cd_value on query
    value: Any = None

    def __repr__(self):
        create_time = (self.created_at.strftime(DATETIME_PRINT_FORMAT)
                       if self.created_at is not None else None)
        update_time = (self.updated_at.strftime(DATETIME_PRINT_FORMAT)
                       if self.updated_at is not None else None)
        return (f'CustomData(id={self.id}, key={self.cd_key}, type={self.cd_type}, ' +
                f'value={self.cd_value}, trade_id={self.ft_trade_id}, created={create_time}, ' +
                f'updated={update_time})')

    @classmethod
    def query_cd(cls, key: Optional[str] = None,
                 trade_id: Optional[int] = None) -> Sequence['_CustomData']:
        """
        Get all CustomData, if trade_id is not specified
        return will be for generic values not tied to a trade
        :param trade_id: id of the Trade
        """
        filters = []
        if trade_id is not None:
            filters.append(_CustomData.ft_trade_id == trade_id)
        if key is not None:
            filters.append(_CustomData.cd_key.ilike(key))

        return _CustomData.session.scalars(select(_CustomData).filter(*filters)).all()


class CustomDataWrapper:
    """
    CustomData middleware class
    Abstracts the database layer away so it becomes optional - which will be necessary to support
    backtesting and hyperopt in the future.
    """

    use_db = True
    custom_data: List[_CustomData] = []
    unserialized_types = ['bool', 'float', 'int', 'str']

    @staticmethod
    def _convert_custom_data(data: _CustomData) -> _CustomData:
        if data.cd_type in CustomDataWrapper.unserialized_types:
            data.value = data.cd_value
            if data.cd_type == 'bool':
                data.value = data.cd_value.lower() == 'true'
            elif data.cd_type == 'int':
                data.value = int(data.cd_value)
            elif data.cd_type == 'float':
                data.value = float(data.cd_value)
        else:
            data.value = json.loads(data.cd_value)
        return data

    @staticmethod
    def reset_custom_data() -> None:
        """
        Resets all key-value pairs. Only active for backtesting mode.
        """
        if not CustomDataWrapper.use_db:
            CustomDataWrapper.custom_data = []

    @staticmethod
    def delete_custom_data(trade_id: int) -> None:
        _CustomData.session.query(_CustomData).filter(_CustomData.ft_trade_id == trade_id).delete()
        _CustomData.session.commit()

    @staticmethod
    def get_custom_data(*, trade_id: int, key: Optional[str] = None) -> List[_CustomData]:

        if CustomDataWrapper.use_db:
            filters = [
                _CustomData.ft_trade_id == trade_id,
            ]
            if key is not None:
                filters.append(_CustomData.cd_key.ilike(key))
            filtered_custom_data = _CustomData.session.scalars(select(_CustomData).filter(
                *filters)).all()

        else:
            filtered_custom_data = [
                data_entry for data_entry in CustomDataWrapper.custom_data
                if (data_entry.ft_trade_id == trade_id)
            ]
            if key is not None:
                filtered_custom_data = [
                    data_entry for data_entry in filtered_custom_data
                    if (data_entry.cd_key.casefold() == key.casefold())
                ]
        return [CustomDataWrapper._convert_custom_data(d) for d in filtered_custom_data]

    @staticmethod
    def set_custom_data(trade_id: int, key: str, value: Any) -> None:

        value_type = type(value).__name__

        if value_type not in CustomDataWrapper.unserialized_types:
            try:
                value_db = json.dumps(value)
            except TypeError as e:
                logger.warning(f"could not serialize {key} value due to {e}")
                return
        else:
            value_db = str(value)

        if trade_id is None:
            trade_id = 0

        custom_data = CustomDataWrapper.get_custom_data(trade_id=trade_id, key=key)
        if custom_data:
            data_entry = custom_data[0]
            data_entry.cd_value = value_db
            data_entry.updated_at = dt_now()
        else:
            data_entry = _CustomData(
                ft_trade_id=trade_id,
                cd_key=key,
                cd_type=value_type,
                cd_value=value_db,
                created_at=dt_now(),
            )
        data_entry.value = value

        if CustomDataWrapper.use_db and value_db is not None:
            _CustomData.session.add(data_entry)
            _CustomData.session.commit()
        else:
            if not custom_data:
                CustomDataWrapper.custom_data.append(data_entry)
            # Existing data will have updated interactively.

1	import json	1✔
2	import logging	1✔
3	from datetime import datetime	1✔
4	from typing import Any, ClassVar, List, Optional, Sequence	1✔
5
6	from sqlalchemy import DateTime, ForeignKey, Integer, String, Text, UniqueConstraint, select	1✔
7	from sqlalchemy.orm import Mapped, mapped_column, relationship	1✔
8
9	from freqtrade.constants import DATETIME_PRINT_FORMAT	1✔
10	from freqtrade.persistence.base import ModelBase, SessionType	1✔
11	from freqtrade.util import dt_now	1✔
12
13
14	logger = logging.getLogger(__name__)	1✔
15
16
17	class _CustomData(ModelBase):	1✔
18	"""
19	CustomData database model
20	Keeps records of metadata as key/value store
21	for trades or global persistent values
22	One to many relationship with Trades:
23	- One trade can have many metadata entries
24	- One metadata entry can only be associated with one Trade
25	"""
26	__tablename__ = 'trade_custom_data'	1✔
27	__allow_unmapped__ = True	1✔
28	session: ClassVar[SessionType]	1✔
29
30	# Uniqueness should be ensured over pair, order_id
31	# its likely that order_id is unique per Pair on some exchanges.
32	__table_args__ = (UniqueConstraint('ft_trade_id', 'cd_key', name="_trade_id_cd_key"),)	1✔
33
34	id = mapped_column(Integer, primary_key=True)	1✔
35	ft_trade_id = mapped_column(Integer, ForeignKey('trades.id'), index=True)	1✔
36
37	trade = relationship("Trade", back_populates="custom_data")	1✔
38
39	cd_key: Mapped[str] = mapped_column(String(255), nullable=False)	1✔
40	cd_type: Mapped[str] = mapped_column(String(25), nullable=False)	1✔
41	cd_value: Mapped[str] = mapped_column(Text, nullable=False)	1✔
42	created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, default=dt_now)	1✔
43	updated_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True)	1✔
44
45	# Empty container value - not persisted, but filled with cd_value on query
46	value: Any = None	1✔
47
48	def __repr__(self):	1✔
49	create_time = (self.created_at.strftime(DATETIME_PRINT_FORMAT)	×
50	if self.created_at is not None else None)
51	update_time = (self.updated_at.strftime(DATETIME_PRINT_FORMAT)	×
52	if self.updated_at is not None else None)
53	return (f'CustomData(id={self.id}, key={self.cd_key}, type={self.cd_type}, ' +	×
54	f'value={self.cd_value}, trade_id={self.ft_trade_id}, created={create_time}, ' +
55	f'updated={update_time})')
56
57	@classmethod	1✔
58	def query_cd(cls, key: Optional[str] = None,	1✔
59	trade_id: Optional[int] = None) -> Sequence['_CustomData']:
60	"""
61	Get all CustomData, if trade_id is not specified
62	return will be for generic values not tied to a trade
63	:param trade_id: id of the Trade
64	"""
65	filters = []	×
66	if trade_id is not None:	×
67	filters.append(_CustomData.ft_trade_id == trade_id)	×
68	if key is not None:	×
69	filters.append(_CustomData.cd_key.ilike(key))	×
70
71	return _CustomData.session.scalars(select(_CustomData).filter(*filters)).all()	×
72
73
74	class CustomDataWrapper:	1✔
75	"""
76	CustomData middleware class
77	Abstracts the database layer away so it becomes optional - which will be necessary to support
78	backtesting and hyperopt in the future.
79	"""
80
81	use_db = True	1✔
82	custom_data: List[_CustomData] = []	1✔
83	unserialized_types = ['bool', 'float', 'int', 'str']	1✔
84
85	@staticmethod	1✔
86	def _convert_custom_data(data: _CustomData) -> _CustomData:	1✔
87	if data.cd_type in CustomDataWrapper.unserialized_types:	1✔
88	data.value = data.cd_value	1✔
89	if data.cd_type == 'bool':	1✔
90	data.value = data.cd_value.lower() == 'true'	1✔
91	elif data.cd_type == 'int':	1✔
92	data.value = int(data.cd_value)	1✔
93	elif data.cd_type == 'float':	1✔
94	data.value = float(data.cd_value)	1✔
95	else:
96	data.value = json.loads(data.cd_value)	1✔
97	return data	1✔
98
99	@staticmethod	1✔
100	def reset_custom_data() -> None:	1✔
101	"""
102	Resets all key-value pairs. Only active for backtesting mode.
103	"""
104	if not CustomDataWrapper.use_db:	1✔
105	CustomDataWrapper.custom_data = []	1✔
106
107	@staticmethod	1✔
108	def delete_custom_data(trade_id: int) -> None:	1✔
109	_CustomData.session.query(_CustomData).filter(_CustomData.ft_trade_id == trade_id).delete()	1✔
110	_CustomData.session.commit()	1✔
111
112	@staticmethod	1✔
113	def get_custom_data(*, trade_id: int, key: Optional[str] = None) -> List[_CustomData]:	1✔
114
115	if CustomDataWrapper.use_db:	1✔
116	filters = [	1✔
117	_CustomData.ft_trade_id == trade_id,
118	]
119	if key is not None:	1✔
120	filters.append(_CustomData.cd_key.ilike(key))	1✔
121	filtered_custom_data = _CustomData.session.scalars(select(_CustomData).filter(	1✔
122	*filters)).all()
123
124	else:
125	filtered_custom_data = [	1✔
126	data_entry for data_entry in CustomDataWrapper.custom_data
127	if (data_entry.ft_trade_id == trade_id)
128	]
129	if key is not None:	1✔
130	filtered_custom_data = [	1✔
131	data_entry for data_entry in filtered_custom_data
132	if (data_entry.cd_key.casefold() == key.casefold())
133	]
134	return [CustomDataWrapper._convert_custom_data(d) for d in filtered_custom_data]	1✔
135
136	@staticmethod	1✔
137	def set_custom_data(trade_id: int, key: str, value: Any) -> None:	1✔
138
139	value_type = type(value).__name__	1✔
140
141	if value_type not in CustomDataWrapper.unserialized_types:	1✔
142	try:	1✔
143	value_db = json.dumps(value)	1✔
144	except TypeError as e:	×
145	logger.warning(f"could not serialize {key} value due to {e}")	×
146	return	×
147	else:
148	value_db = str(value)	1✔
149
150	if trade_id is None:	1✔
151	trade_id = 0	×
152
153	custom_data = CustomDataWrapper.get_custom_data(trade_id=trade_id, key=key)	1✔
154	if custom_data:	1✔
155	data_entry = custom_data[0]	1✔
156	data_entry.cd_value = value_db	1✔
157	data_entry.updated_at = dt_now()	1✔
158	else:
159	data_entry = _CustomData(	1✔
160	ft_trade_id=trade_id,
161	cd_key=key,
162	cd_type=value_type,
163	cd_value=value_db,
164	created_at=dt_now(),
165	)
166	data_entry.value = value	1✔
167
168	if CustomDataWrapper.use_db and value_db is not None:	1✔
169	_CustomData.session.add(data_entry)	1✔
170	_CustomData.session.commit()	1✔
171	else:
172	if not custom_data:	1✔
173	CustomDataWrapper.custom_data.append(data_entry)	1✔
174	# Existing data will have updated interactively.

freqtrade / freqtrade / 9394559170

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous