Skip to content

Commit

Permalink
Cleaning up
Browse files Browse the repository at this point in the history
  • Loading branch information
@4TT1L4
4TT1L4 committed Aug 4, 2024
1 parent 0f2e9d1 commit ac01ec9
Showing 1 changed file with 74 additions and 190 deletions.
264 changes: 74 additions & 190 deletions src/strategies/combined_rsi_bollinger_strategy.py
View file
Original file line number Diff line number Diff line change
@@ -1,27 +1,34 @@
import logging
import math
import time
from datetime import datetime
from typing import Any, Dict, List, Optional
from typing import Dict, Optional

from talipp.indicators import BB, RSI

from api import Api, ApiException
from api import Api
from src.data_extraction.fear_and_greed_index_web_scraper import (
FearAndGreedIndexWebScraper,
)
from src.models.candlestick import Candlestick
from src.utils.logger_utils import LoggerUtils
from src.utils.market_maker import MarketMaker


# pylint: disable=invalid-name
class combined_rsi_bollinger_strategy:
"""
A trading strategy combining RSI and Bollinger Bands.
A trading strategy combining RSI, Bollinger Bands, and optionally Fear & Greed Index.
This class implements a trading strategy that uses RSI and Bollinger Bands indicators
to make buy and sell decisions in a given market.
This class implements a trading strategy that uses RSI and Bollinger Bands indicators,
along with an optional Fear & Greed Index to make buy and sell decisions in a given market.
Attributes:
api_client (Api): The API client for market interactions.
logger (logging.Logger): Logger for outputting information and errors.
market_maker (MarketMaker): Handles order placement and management.
rsi (RSI): RSI indicator instance.
bb (BB): Bollinger Bands indicator instance.
fgis (Optional[FearGreedIndexScraper]): Fear & Greed Index scraper, if enabled.
"""

def __init__(self, api_client: Api, config: Dict[str, any], logger: logging.Logger):
Expand All @@ -37,24 +44,14 @@ def __init__(self, api_client: Api, config: Dict[str, any], logger: logging.Logg
ValueError: If the configuration is invalid.
"""
logger.info(" > init: combined_rsi_bollinger_strategy instance created.")

logger.info("========================================================================")
logger.info(" ")
logger.info(" ⚠️ WARNING! ⚠️ ")
logger.info(" ")
logger.info(" THIS IS ONLY A PROOF-OF-CONCEPT EXAMPLE STRATEGY IMPLEMENTATION. ")
logger.info(" ")
logger.info(" IT IS ONLY INTENDED AS IMPLEMENTATION REFERENCE FOR TRADING STRATEGIES.")
logger.info(" ")
logger.info(" THIS IMPLEMENTATION IS NOT PRODUCTION-READY. ")
logger.info(" ")
logger.info("========================================================================")
LoggerUtils.log_warning(logger)

self.api_client = api_client
self.logger = logger
self.initialized = False
self.last_candle: Optional[Any] = None
self.last_candle: Optional[Candlestick] = None
self.last_execution_time: Optional[datetime] = None
self.cached_fear_and_greed_index = None

try:
self.validate_config(config)
Expand All @@ -77,6 +74,8 @@ def _initialize_strategy_parameters(self, config: Dict[str, any]) -> None:
self.base_asset = config["BASE_ASSET"]
self.target_asset = config["TARGET_ASSET"]
self.market = f"{self.base_asset}_{self.target_asset}"
self.use_fear_and_greed = config.get("USE_FEAR_AND_GREED", False)
self.fear_and_greed_index_threshold = int(config["FEAR_AND_GREED_INDEX_THRESHOLD"])

def _log_strategy_configuration(self) -> None:
"""Log the strategy configuration."""
Expand All @@ -90,11 +89,16 @@ def _log_strategy_configuration(self) -> None:
self.logger.info(f" > rsi_oversold : {self.rsi_oversold}")
self.logger.info(f" > bb_period : {self.bb_period}")
self.logger.info(f" > bb_std_dev : {self.bb_std_dev}")
self.logger.info(f" > use_fear_and_greed : {self.use_fear_and_greed}")
if self.use_fear_and_greed:
self.logger.info(f" > fear_greed_threshold: {self.fear_and_greed_index_threshold}")

def _initialize_indicators(self, config: Dict[str, any]) -> None:
"""Initialize strategy indicators and components."""
self.rsi = RSI(self.rsi_period)
self.bb = BB(self.bb_period, self.bb_std_dev)
self.market_maker = MarketMaker(self.api_client, config, self.logger)
self.fgis = FearAndGreedIndexWebScraper(self.logger) if self.use_fear_and_greed else None

@staticmethod
def validate_config(config: Dict[str, any]) -> None:
Expand All @@ -109,7 +113,7 @@ def validate_config(config: Dict[str, any]) -> None:
"""
required_fields = [
"POSITION_SIZE_LOVELACES", "RSI_PERIOD", "RSI_OVERBOUGHT", "RSI_OVERSOLD",
"BB_PERIOD", "BB_STD_DEV", "BASE_ASSET", "TARGET_ASSET"
"BB_PERIOD", "BB_STD_DEV", "BASE_ASSET", "TARGET_ASSET", "FEAR_AND_GREED_INDEX_THRESHOLD"
]

for field in required_fields:
Expand All @@ -118,14 +122,14 @@ def validate_config(config: Dict[str, any]) -> None:

# Validate types and ranges
try:
assert float(config["POSITION_SIZE_LOVELACES"]) > 0, \
"POSITION_SIZE_LOVELACES must be positive"
assert float(config["POSITION_SIZE_LOVELACES"]) > 0, "POSITION_SIZE_LOVELACES must be positive"
assert 2 <= int(config["RSI_PERIOD"]) <= 100, "RSI_PERIOD must be between 2 and 100"
assert 50 <= float(config["RSI_OVERBOUGHT"]) <= 100, \
"RSI_OVERBOUGHT must be between 50 and 100"
assert 50 <= float(config["RSI_OVERBOUGHT"]) <= 100, "RSI_OVERBOUGHT must be between 50 and 100"
assert 0 <= float(config["RSI_OVERSOLD"]) <= 50, "RSI_OVERSOLD must be between 0 and 50"
assert 2 <= int(config["BB_PERIOD"]) <= 100, "BB_PERIOD must be between 2 and 100"
assert 0 < float(config["BB_STD_DEV"]) <= 5, "BB_STD_DEV must be between 0 and 5"
assert 0 <= int(config["FEAR_AND_GREED_INDEX_THRESHOLD"]) <= 100, \
"FEAR_AND_GREED_INDEX_THRESHOLD must be between 0 and 100"
assert isinstance(config["BASE_ASSET"], str) and config["BASE_ASSET"].strip(), \
"BASE_ASSET should be a non-empty string"
assert isinstance(config["TARGET_ASSET"], str) and config["TARGET_ASSET"].strip(), \
Expand All @@ -135,22 +139,29 @@ def validate_config(config: Dict[str, any]) -> None:
except AssertionError as e:
raise ValueError(f"Configuration validation failed: {str(e)}") from e

def process_candle(self, candle) -> None:
def get_fear_and_greed_index(self) -> Optional[int]:
"""
Get the current Fear and Greed Index value.
Returns:
Optional[int]: The Fear and Greed Index value, or None if not available or not used.
"""
if not self.use_fear_and_greed or self.fgis is None:
return None
return self.fgis.get_index_value() or self.fear_and_greed_index_threshold

def process_candle(self, candle: Candlestick) -> None:
"""
Process a new candle and make trading decisions.
Args:
candle (Candle): The new candle to process.
"""
if self.initialized:
self.logger.info(
f" > processing candle - timestamp: {candle.timestamp} \
- base_close: {candle.base_close}"
)
self.logger.info(f" > processing candle - timestamp: {candle.timestamp} - base_close: {candle.base_close}")
else:
self.logger.info(
f" > processing init candle - timestamp: {candle.timestamp} \
- base_close: {candle.base_close}"
f" > processing init candle - timestamp: {candle.timestamp} - base_close: {candle.base_close}"
)

if self.last_candle and self.last_candle.timestamp == candle.timestamp:
Expand Down Expand Up @@ -184,43 +195,58 @@ def _log_indicator_values(self, value: float) -> None:
current_rsi = self.rsi[-1]
current_bb = self.bb[-1]
middle_band = (current_bb.ub + current_bb.lb) / 2
fear_and_greed_index = None
if self.initialized:
# Prevent unnecessary calls to external website.
fear_and_greed_index = self.get_fear_and_greed_index()
self.cached_fear_and_greed_index = fear_and_greed_index

self.logger.info(f" RSI: {current_rsi:.2f}")
self.logger.info(f" BB: Lower {current_bb.lb:.2f}, \
Middle {middle_band:.2f}, Upper {current_bb.ub:.2f}")
self.logger.info(f" BB: Lower {current_bb.lb:.2f}, Middle {middle_band:.2f}, Upper {current_bb.ub:.2f}")
if fear_and_greed_index is not None:
self.logger.info(f" Fear & Greed Index: {fear_and_greed_index}")
else:
self.logger.info(" Fear & Greed Index: Not available")

def _execute_trading_logic(self, value: float) -> None:
"""Execute the trading logic based on indicator values."""
current_rsi = self.rsi[-1]
current_bb = self.bb[-1]
fear_and_greed_index = self.cached_fear_and_greed_index or self.get_fear_and_greed_index()

buy_signal = (current_rsi < self.rsi_oversold and value <= current_bb.lb)
sell_signal = (current_rsi > self.rsi_overbought and value >= current_bb.ub)

if fear_and_greed_index is not None:
buy_signal = buy_signal and (fear_and_greed_index < self.fear_and_greed_index_threshold)
sell_signal = sell_signal and (fear_and_greed_index > 100 - self.fear_and_greed_index_threshold)

if buy_signal:
self._handle_buy_signal()
self._handle_buy_signal(fear_and_greed_index)
elif sell_signal:
self._handle_sell_signal()
self._handle_sell_signal(fear_and_greed_index)
else:
self.logger.info(" -> No clear signal or conflicting indicators. Holding position.")

self.log_orders()
self.market_maker.log_orders()

def _handle_buy_signal(self) -> None:
def _handle_buy_signal(self, fear_and_greed_index: Optional[int]) -> None:
"""Handle a buy signal."""
self.logger.info(" -> Strong BUY signal: RSI oversold, price below lower BB")
self.cancel_sell_orders()
if not self.get_buy_orders():
self.place_buy_order(self.last_candle.base_close)
self.logger.info(" -> Strong BUY signal: RSI oversold, price below lower BB" +
(f", high fear (index: {fear_and_greed_index})" if fear_and_greed_index is not None else ""))
self.market_maker.cancel_sell_orders()
if not self.market_maker.get_buy_orders():
self.market_maker.place_buy_order(self.last_candle.base_close)
else:
self.logger.info(" > Already placed BUY order. Nothing to do.")

def _handle_sell_signal(self) -> None:
def _handle_sell_signal(self, fear_and_greed_index: Optional[int]) -> None:
"""Handle a sell signal."""
self.logger.info(" -> Strong SELL signal: RSI overbought, price above upper BB")
self.cancel_buy_orders()
if not self.get_sell_orders():
self.place_sell_order(self.last_candle.base_close)
self.logger.info(" -> Strong SELL signal: RSI overbought, price above upper BB" +
(f", high greed (index: {fear_and_greed_index})" if fear_and_greed_index is not None else ""))
self.market_maker.cancel_buy_orders()
if not self.market_maker.get_sell_orders():
self.market_maker.place_sell_order(self.last_candle.base_close)
else:
self.logger.info(" > Already placed SELL order. Nothing to do.")

Expand Down Expand Up @@ -273,152 +299,10 @@ def _initialize_strategy(self, api_client: Api) -> None:
limit=max(self.rsi_period, self.bb_period) * 5
)
for candle in candles[:-1]:
self.logger.info(
"--------------------------------------------------------------------------------"
)
self.logger.info("--------------------------------------------------------------------------------")
self.process_candle(candle)
time.sleep(1)
self.logger.info(" > [OK] Initialized.")
self.logger.info("========================================================================")
self.initialized = True
self.last_candle = None

def place_buy_order(self, price: float) -> None:
"""
Place a buy order at the specified price.
Args:
price (float): The price at which to place the buy order.
"""
self.logger.info(" ⚙️ Placing BUY order...")

try:
balance_available = int(self.api_client.get_balances().get(self.base_asset, 0))
self.logger.debug(f" > balance_available : {balance_available}")
self.logger.debug(f" > self.position_size: {self.position_size}")

order_size = min(self.position_size, balance_available)
if not order_size:
self.logger.info(" ⚠️ Insufficient balance to place BUY order! ⚠️")
return

offered_amount = int(math.floor(order_size))

self.logger.info(f" > Place BUY order: {offered_amount} at price {price}...")
response = self.api_client.place_order(
offered_amount=f"{offered_amount}",
offered_token=self.base_asset,
price_token=self.target_asset,
price_amount=f"{int(math.floor(offered_amount / price))}"
)
self.logger.info(f" > [OK] PLACED NEW BUY ORDER: {response.order_ref}")
# pylint: disable=broad-exception-caught
except Exception as e:
self.logger.error(f" > ⚠️ [FAILED] Could not place BUY order: {str(e)} ⚠️")
self.logger.exception(" > Exception details: ")

def place_sell_order(self, price: float) -> None:
"""
Place a sell order at the specified price.
Args:
price (float): The price at which to place the sell order.
"""
self.logger.info(" ⚙️ Placing SELL order...")

try:
balance_available = int(self.api_client.get_balances().get(self.target_asset, 0))
self.logger.info(f" > balance_available : {balance_available}")
order_size = min(self.position_size / price, balance_available)
self.logger.info(f" > order_size : {order_size}")
self.logger.info(f" > price : {price}")
if not order_size:
self.logger.info("⚠️ Insufficient balance to place SELL order! ⚠️")
return

self.logger.info(f" > Place SELL order: {order_size} at price {price}...")
response = self.api_client.place_order(
offered_amount=f"{int(math.floor(order_size))}",
offered_token=self.target_asset,
price_token=self.base_asset,
price_amount=f"{int(math.floor(order_size * price))}"
)
self.logger.info(f" > [OK] PLACED NEW SELL ORDER: {response.order_ref}")
# pylint: disable=broad-exception-caught
except Exception as e:
self.logger.error(f" > ⚠️ [FAILED] Could not place SELL order: {str(e)} ⚠️")
self.logger.exception(" > Exception details: ")

def get_buy_orders(self) -> List[Any]:
"""
Get all active buy orders for the current market.
Returns:
List[Any]: A list of active buy orders.
"""
own_orders = self.api_client.get_own_orders(self.market)
return own_orders.bids

def get_sell_orders(self) -> List[Any]:
"""
Get all active sell orders for the current market.
Returns:
List[Any]: A list of active sell orders.
"""
own_orders = self.api_client.get_own_orders(self.market)
return own_orders.asks

def cancel_buy_orders(self) -> None:
"""Cancel all active buy orders for the current market."""
self.logger.info(" > Cancel all BUY orders...")
self.cancel_orders("bid")
self.logger.info(" > [OK] Canceled all BUY orders.")

def cancel_sell_orders(self) -> None:
"""Cancel all active sell orders for the current market."""
self.logger.info(" > Cancel all SELL orders...")
self.cancel_orders("ask")
self.logger.info(" > [OK] Canceled all SELL orders.")

def cancel_orders(self, side: str) -> None:
"""
Cancel all orders of a specific side (buy or sell).
Args:
side (str): The side of orders to cancel ("ask" for sell, "bid" for buy).
"""
while True:
orders = self.get_sell_orders() if side == "ask" else self.get_buy_orders()

if not orders:
return

self.logger.info(f" Remaining {side} orders: {len(orders)}.")

order = orders[0]
try:
self.logger.info(f" ⚙️ Canceling order: {order.output_reference}")
self.api_client.cancel_order(order.output_reference)
self.logger.info(f" > [OK] Canceled order: {order.output_reference}")
except ApiException as e:
self.logger.error(
f" > ⚠️ [FAILED] could not cancel order: {order.output_reference} ⚠️"
)
self.logger.exception(f" > Exception: {str(e)}")

def log_orders(self) -> None:
"""Log all active orders for the current market."""
own_orders = self.api_client.get_own_orders(self.market)

self.logger.info(" ON-CHAIN ORDERS:")

if not own_orders.asks and not own_orders.bids:
self.logger.info(" > No orders.")
return

for sell_order in own_orders.asks:
self.logger.info(f" > SELL: {sell_order.output_reference}")

for buy_order in own_orders.bids:
self.logger.info(f" > BUY: {buy_order.output_reference}")
self.last_candle = None

0 comments on commit ac01ec9

Please sign in to comment.