4131167254

Build Type

push

github-actions

Committed by GitHub

Commit Message

Merge pull request #7983 from stash86/bt-metrics

Run Details

16866 of 17748 relevant lines covered (95.03%)

0.95 hits per line

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

92.86

/freqtrade/freqai/RL/Base5ActionRLEnv.py

import logging
from enum import Enum

from gym import spaces

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


logger = logging.getLogger(__name__)


class Actions(Enum):
    Neutral = 0
    Long_enter = 1
    Long_exit = 2
    Short_enter = 3
    Short_exit = 4


class Base5ActionRLEnv(BaseEnvironment):
    """
    Base class for a 5 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])

        trade_type = None
        if self.is_tradesignal(action):
            """
            Action: Neutral, position: Long ->  Close Long
            Action: Neutral, position: Short -> Close Short

            Action: Long, position: Neutral -> Open Long
            Action: Long, position: Short -> Close Short and Open Long

            Action: Short, position: Neutral -> Open Short
            Action: Short, position: Long -> Close Long and Open Short
            """

            if action == Actions.Neutral.value:
                self._position = Positions.Neutral
                trade_type = "neutral"
                self._last_trade_tick = None
            elif action == Actions.Long_enter.value:
                self._position = Positions.Long
                trade_type = "long"
                self._last_trade_tick = self._current_tick
            elif action == Actions.Short_enter.value:
                self._position = Positions.Short
                trade_type = "short"
                self._last_trade_tick = self._current_tick
            elif action == Actions.Long_exit.value:
                self._update_total_profit()
                self._position = Positions.Neutral
                trade_type = "neutral"
                self._last_trade_tick = None
            elif action == Actions.Short_exit.value:
                self._update_total_profit()
                self._position = Positions.Neutral
                trade_type = "neutral"
                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})

        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()

        self._update_history(info)

        return observation, step_reward, self._done, info

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

    def _is_valid(self, action: int) -> bool:
        # trade signal
        """
        Determine if the signal is valid.
        e.g.: agent wants a Actions.Long_exit while it is in a Positions.short
        """
        # Agent should only try to exit if it is in position
        if action in (Actions.Short_exit.value, Actions.Long_exit.value):
            if self._position not in (Positions.Short, Positions.Long):
                return False

        # Agent should only try to enter if it is not in position
        if action in (Actions.Short_enter.value, Actions.Long_enter.value):
            if self._position != Positions.Neutral:
                return False

        return True

1	import logging	1✔
2	from enum import Enum	1✔
3
4	from gym 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	Long_enter = 1	1✔
15	Long_exit = 2	1✔
16	Short_enter = 3	1✔
17	Short_exit = 4	1✔
18
19
20	class Base5ActionRLEnv(BaseEnvironment):	1✔
21	"""
22	Base class for a 5 action environment
23	"""
24	def __init__(self, **kwargs):	1✔
25	super().__init__(**kwargs)	1✔
26	self.actions = Actions	1✔
27
28	def set_action_space(self):	1✔
29	self.action_space = spaces.Discrete(len(Actions))	1✔
30
31	def step(self, action: int):	1✔
32	"""
33	Logic for a single step (incrementing one candle in time)
34	by the agent
35	:param: action: int = the action type that the agent plans
36	to take for the current step.
37	:returns:
38	observation = current state of environment
39	step_reward = the reward from `calculate_reward()`
40	_done = if the agent "died" or if the candles finished
41	info = dict passed back to openai gym lib
42	"""
43	self._done = False	1✔
44	self._current_tick += 1	1✔
45
46	if self._current_tick == self._end_tick:	1✔
47	self._done = True	1✔
48
49	self._update_unrealized_total_profit()	1✔
50	step_reward = self.calculate_reward(action)	1✔
51	self.total_reward += step_reward	1✔
52	self.tensorboard_log(self.actions._member_names_[action])	1✔
53
54	trade_type = None	1✔
55	if self.is_tradesignal(action):	1✔
56	"""
57	Action: Neutral, position: Long -> Close Long
58	Action: Neutral, position: Short -> Close Short
59
60	Action: Long, position: Neutral -> Open Long
61	Action: Long, position: Short -> Close Short and Open Long
62
63	Action: Short, position: Neutral -> Open Short
64	Action: Short, position: Long -> Close Long and Open Short
65	"""
66
67	if action == Actions.Neutral.value:	1✔
68	self._position = Positions.Neutral	×
69	trade_type = "neutral"	×
70	self._last_trade_tick = None	×
71	elif action == Actions.Long_enter.value:	1✔
72	self._position = Positions.Long	1✔
73	trade_type = "long"	1✔
74	self._last_trade_tick = self._current_tick	1✔
75	elif action == Actions.Short_enter.value:	1✔
76	self._position = Positions.Short	1✔
77	trade_type = "short"	1✔
78	self._last_trade_tick = self._current_tick	1✔
79	elif action == Actions.Long_exit.value:	1✔
80	self._update_total_profit()	1✔
81	self._position = Positions.Neutral	1✔
82	trade_type = "neutral"	1✔
83	self._last_trade_tick = None	1✔
84	elif action == Actions.Short_exit.value:	1✔
85	self._update_total_profit()	1✔
86	self._position = Positions.Neutral	1✔
87	trade_type = "neutral"	1✔
88	self._last_trade_tick = None	1✔
89	else:
90	print("case not defined")	×
91
92	if trade_type is not None:	1✔
93	self.trade_history.append(	1✔
94	{'price': self.current_price(), 'index': self._current_tick,
95	'type': trade_type})
96
97	if (self._total_profit < self.max_drawdown or	1✔
98	self._total_unrealized_profit < self.max_drawdown):
99	self._done = True	×
100
101	self._position_history.append(self._position)	1✔
102
103	info = dict(	1✔
104	tick=self._current_tick,
105	action=action,
106	total_reward=self.total_reward,
107	total_profit=self._total_profit,
108	position=self._position.value,
109	trade_duration=self.get_trade_duration(),
110	current_profit_pct=self.get_unrealized_profit()
111	)
112
113	observation = self._get_observation()	1✔
114
115	self._update_history(info)	1✔
116
117	return observation, step_reward, self._done, info	1✔
118
119	def is_tradesignal(self, action: int) -> bool:	1✔
120	"""
121	Determine if the signal is a trade signal
122	e.g.: agent wants a Actions.Long_exit while it is in a Positions.short
123	"""
124	return not ((action == Actions.Neutral.value and self._position == Positions.Neutral) or	1✔
125	(action == Actions.Neutral.value and self._position == Positions.Short) or
126	(action == Actions.Neutral.value and self._position == Positions.Long) or
127	(action == Actions.Short_enter.value and self._position == Positions.Short) or
128	(action == Actions.Short_enter.value and self._position == Positions.Long) or
129	(action == Actions.Short_exit.value and self._position == Positions.Long) or
130	(action == Actions.Short_exit.value and self._position == Positions.Neutral) or
131	(action == Actions.Long_enter.value and self._position == Positions.Long) or
132	(action == Actions.Long_enter.value and self._position == Positions.Short) or
133	(action == Actions.Long_exit.value and self._position == Positions.Short) or
134	(action == Actions.Long_exit.value and self._position == Positions.Neutral))
135
136	def _is_valid(self, action: int) -> bool:	1✔
137	# trade signal
138	"""
139	Determine if the signal is valid.
140	e.g.: agent wants a Actions.Long_exit while it is in a Positions.short
141	"""
142	# Agent should only try to exit if it is in position
143	if action in (Actions.Short_exit.value, Actions.Long_exit.value):	1✔
144	if self._position not in (Positions.Short, Positions.Long):	1✔
145	return False	1✔
146
147	# Agent should only try to enter if it is not in position
148	if action in (Actions.Short_enter.value, Actions.Long_enter.value):	1✔
149	if self._position != Positions.Neutral:	1✔
150	return False	1✔
151
152	return True	1✔

freqtrade / freqtrade / 4131167254

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