/
bollinger_v1.py
101 lines (89 loc) · 4.05 KB
/
bollinger_v1.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
from typing import List
import pandas_ta as ta # noqa: F401
from pydantic import Field, validator
from hummingbot.client.config.config_data_types import ClientFieldData
from hummingbot.data_feed.candles_feed.data_types import CandlesConfig
from hummingbot.strategy_v2.controllers.directional_trading_controller_base import (
DirectionalTradingControllerBase,
DirectionalTradingControllerConfigBase,
)
class BollingerV1ControllerConfig(DirectionalTradingControllerConfigBase):
controller_name = "bollinger_v1"
candles_config: List[CandlesConfig] = []
candles_connector: str = Field(
default=None,
client_data=ClientFieldData(
prompt_on_new=True,
prompt=lambda mi: "Enter the connector for the candles data, leave empty to use the same exchange as the connector: ", )
)
candles_trading_pair: str = Field(
default=None,
client_data=ClientFieldData(
prompt_on_new=True,
prompt=lambda mi: "Enter the trading pair for the candles data, leave empty to use the same trading pair as the connector: ", )
)
interval: str = Field(
default="3m",
client_data=ClientFieldData(
prompt=lambda mi: "Enter the candle interval (e.g., 1m, 5m, 1h, 1d): ",
prompt_on_new=False))
bb_length: int = Field(
default=100,
client_data=ClientFieldData(
prompt=lambda mi: "Enter the Bollinger Bands length: ",
prompt_on_new=True))
bb_std: float = Field(
default=2.0,
client_data=ClientFieldData(
prompt=lambda mi: "Enter the Bollinger Bands standard deviation: ",
prompt_on_new=False))
bb_long_threshold: float = Field(
default=0.0,
client_data=ClientFieldData(
prompt=lambda mi: "Enter the Bollinger Bands long threshold: ",
prompt_on_new=True))
bb_short_threshold: float = Field(
default=1.0,
client_data=ClientFieldData(
prompt=lambda mi: "Enter the Bollinger Bands short threshold: ",
prompt_on_new=True))
@validator("candles_connector", pre=True, always=True)
def set_candles_connector(cls, v, values):
if v is None or v == "":
return values.get("connector_name")
return v
@validator("candles_trading_pair", pre=True, always=True)
def set_candles_trading_pair(cls, v, values):
if v is None or v == "":
return values.get("trading_pair")
return v
class BollingerV1Controller(DirectionalTradingControllerBase):
def __init__(self, config: BollingerV1ControllerConfig, *args, **kwargs):
self.config = config
self.max_records = self.config.bb_length
if len(self.config.candles_config) == 0:
self.config.candles_config = [CandlesConfig(
connector=config.candles_connector,
trading_pair=config.candles_trading_pair,
interval=config.interval,
max_records=self.max_records
)]
super().__init__(config, *args, **kwargs)
async def update_processed_data(self):
df = self.market_data_provider.get_candles_df(connector_name=self.config.candles_connector,
trading_pair=self.config.candles_trading_pair,
interval=self.config.interval,
max_records=self.max_records)
# Add indicators
df.ta.bbands(length=self.config.bb_length, std=self.config.bb_std, append=True)
bbp = df[f"BBP_{self.config.bb_length}_{self.config.bb_std}"]
# Generate signal
long_condition = bbp < self.config.bb_long_threshold
short_condition = bbp > self.config.bb_short_threshold
# Generate signal
df["signal"] = 0
df.loc[long_condition, "signal"] = 1
df.loc[short_condition, "signal"] = -1
# Update processed data
self.processed_data["signal"] = df["signal"].iloc[-1]
self.processed_data["features"] = df