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

98.39

/freqtrade/freqai/RL/Base3ActionRLEnv.py

import logging
from enum import Enum

from gymnasium import spaces

from freqtrade.freqai.RL.BaseEnvironment import BaseEnvironment, Positions


logger = logging.getLogger(__name__)


class Actions(Enum):
    Neutral = 0
    Buy = 1
    Sell = 2


class Base3ActionRLEnv(BaseEnvironment):
    """
    Base class for a 3 action environment
    """
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.actions = Actions

    def set_action_space(self):
        self.action_space = spaces.Discrete(len(Actions))

    def step(self, action: int):
        """
        Logic for a single step (incrementing one candle in time)
        by the agent
        :param: action: int = the action type that the agent plans
            to take for the current step.
        :returns:
            observation = current state of environment
            step_reward = the reward from `calculate_reward()`
            _done = if the agent "died" or if the candles finished
            info = dict passed back to openai gym lib
        """
        self._done = False
        self._current_tick += 1

        if self._current_tick == self._end_tick:
            self._done = True

        self._update_unrealized_total_profit()
        step_reward = self.calculate_reward(action)
        self.total_reward += step_reward
        self.tensorboard_log(self.actions._member_names_[action], category="actions")

        trade_type = None
        if self.is_tradesignal(action):
            if action == Actions.Buy.value:
                if self._position == Positions.Short:
                    self._update_total_profit()
                self._position = Positions.Long
                trade_type = "long"
                self._last_trade_tick = self._current_tick
            elif action == Actions.Sell.value and self.can_short:
                if self._position == Positions.Long:
                    self._update_total_profit()
                self._position = Positions.Short
                trade_type = "short"
                self._last_trade_tick = self._current_tick
            elif action == Actions.Sell.value and not self.can_short:
                self._update_total_profit()
                self._position = Positions.Neutral
                trade_type = "exit"
                self._last_trade_tick = None
            else:
                print("case not defined")

            if trade_type is not None:
                self.trade_history.append(
                    {'price': self.current_price(), 'index': self._current_tick,
                     'type': trade_type, 'profit': self.get_unrealized_profit()})

        if (self._total_profit < self.max_drawdown or
                self._total_unrealized_profit < self.max_drawdown):
            self._done = True

        self._position_history.append(self._position)

        info = dict(
            tick=self._current_tick,
            action=action,
            total_reward=self.total_reward,
            total_profit=self._total_profit,
            position=self._position.value,
            trade_duration=self.get_trade_duration(),
            current_profit_pct=self.get_unrealized_profit()
        )

        observation = self._get_observation()

        # user can play with time if they want
        truncated = False

        self._update_history(info)

        return observation, step_reward, self._done, truncated, info

    def is_tradesignal(self, action: int) -> bool:
        """
        Determine if the signal is a trade signal
        e.g.: agent wants a Actions.Buy while it is in a Positions.short
        """
        return (
            (action == Actions.Buy.value and self._position == Positions.Neutral)
            or (action == Actions.Sell.value and self._position == Positions.Long)
            or (action == Actions.Sell.value and self._position == Positions.Neutral
                and self.can_short)
            or (action == Actions.Buy.value and self._position == Positions.Short
                and self.can_short)
        )

    def _is_valid(self, action: int) -> bool:
        """
        Determine if the signal is valid.
        e.g.: agent wants a Actions.Sell while it is in a Positions.Long
        """
        if self.can_short:
            return action in [Actions.Buy.value, Actions.Sell.value, Actions.Neutral.value]
        else:
            if action == Actions.Sell.value and self._position != Positions.Long:
                return False
            return True

1	import logging	1✔
2	from enum import Enum	1✔
3
4	from gymnasium import spaces	1✔
5
6	from freqtrade.freqai.RL.BaseEnvironment import BaseEnvironment, Positions	1✔
7
8
9	logger = logging.getLogger(__name__)	1✔
10
11
12	class Actions(Enum):	1✔
13	Neutral = 0	1✔
14	Buy = 1	1✔
15	Sell = 2	1✔
16
17
18	class Base3ActionRLEnv(BaseEnvironment):	1✔
19	"""
20	Base class for a 3 action environment
21	"""
22	def __init__(self, **kwargs):	1✔
23	super().__init__(**kwargs)	1✔
24	self.actions = Actions	1✔
25
26	def set_action_space(self):	1✔
27	self.action_space = spaces.Discrete(len(Actions))	1✔
28
29	def step(self, action: int):	1✔
30	"""
31	Logic for a single step (incrementing one candle in time)
32	by the agent
33	:param: action: int = the action type that the agent plans
34	to take for the current step.
35	:returns:
36	observation = current state of environment
37	step_reward = the reward from `calculate_reward()`
38	_done = if the agent "died" or if the candles finished
39	info = dict passed back to openai gym lib
40	"""
41	self._done = False	1✔
42	self._current_tick += 1	1✔
43
44	if self._current_tick == self._end_tick:	1✔
45	self._done = True	1✔
46
47	self._update_unrealized_total_profit()	1✔
48	step_reward = self.calculate_reward(action)	1✔
49	self.total_reward += step_reward	1✔
50	self.tensorboard_log(self.actions._member_names_[action], category="actions")	1✔
51
52	trade_type = None	1✔
53	if self.is_tradesignal(action):	1✔
54	if action == Actions.Buy.value:	1✔
55	if self._position == Positions.Short:	1✔
56	self._update_total_profit()	1✔
57	self._position = Positions.Long	1✔
58	trade_type = "long"	1✔
59	self._last_trade_tick = self._current_tick	1✔
60	elif action == Actions.Sell.value and self.can_short:	1✔
61	if self._position == Positions.Long:	1✔
62	self._update_total_profit()	1✔
63	self._position = Positions.Short	1✔
64	trade_type = "short"	1✔
65	self._last_trade_tick = self._current_tick	1✔
66	elif action == Actions.Sell.value and not self.can_short:	1✔
67	self._update_total_profit()	1✔
68	self._position = Positions.Neutral	1✔
69	trade_type = "exit"	1✔
70	self._last_trade_tick = None	1✔
71	else:
72	print("case not defined")	×
73
74	if trade_type is not None:	1✔
75	self.trade_history.append(	1✔
76	{'price': self.current_price(), 'index': self._current_tick,
77	'type': trade_type, 'profit': self.get_unrealized_profit()})
78
79	if (self._total_profit < self.max_drawdown or	1✔
80	self._total_unrealized_profit < self.max_drawdown):
81	self._done = True	1✔
82
83	self._position_history.append(self._position)	1✔
84
85	info = dict(	1✔
86	tick=self._current_tick,
87	action=action,
88	total_reward=self.total_reward,
89	total_profit=self._total_profit,
90	position=self._position.value,
91	trade_duration=self.get_trade_duration(),
92	current_profit_pct=self.get_unrealized_profit()
93	)
94
95	observation = self._get_observation()	1✔
96
97	# user can play with time if they want
98	truncated = False	1✔
99
100	self._update_history(info)	1✔
101
102	return observation, step_reward, self._done, truncated, info	1✔
103
104	def is_tradesignal(self, action: int) -> bool:	1✔
105	"""
106	Determine if the signal is a trade signal
107	e.g.: agent wants a Actions.Buy while it is in a Positions.short
108	"""
109	return (	1✔
110	(action == Actions.Buy.value and self._position == Positions.Neutral)
111	or (action == Actions.Sell.value and self._position == Positions.Long)
112	or (action == Actions.Sell.value and self._position == Positions.Neutral
113	and self.can_short)
114	or (action == Actions.Buy.value and self._position == Positions.Short
115	and self.can_short)
116	)
117
118	def _is_valid(self, action: int) -> bool:	1✔
119	"""
120	Determine if the signal is valid.
121	e.g.: agent wants a Actions.Sell while it is in a Positions.Long
122	"""
123	if self.can_short:	1✔
124	return action in [Actions.Buy.value, Actions.Sell.value, Actions.Neutral.value]	1✔
125	else:
126	if action == Actions.Sell.value and self._position != Positions.Long:	1✔
127	return False	1✔
128	return True	1✔

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